feat: add admin API endpoints for inventory transaction history

Adds admin endpoints for viewing transaction history across all vendors:
- GET /admin/inventory/transactions - paginated cross-vendor list
- GET /admin/inventory/transactions/stats - platform-wide statistics

Includes vendor details in transaction items and transaction counts
by type for dashboard displays.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2026-01-01 18:22:51 +01:00
parent 159243066c
commit 55c1a43f56
3 changed files with 220 additions and 0 deletions

View File

@@ -22,6 +22,7 @@ from app.api.deps import get_current_admin_api
from app.core.database import get_db
from app.services.inventory_import_service import inventory_import_service
from app.services.inventory_service import inventory_service
from app.services.inventory_transaction_service import inventory_transaction_service
from models.database.user import User
from models.schema.inventory import (
AdminInventoryAdjust,
@@ -29,7 +30,10 @@ from models.schema.inventory import (
AdminInventoryListResponse,
AdminInventoryLocationsResponse,
AdminInventoryStats,
AdminInventoryTransactionItem,
AdminInventoryTransactionListResponse,
AdminLowStockItem,
AdminTransactionStatsResponse,
AdminVendorsWithInventoryResponse,
InventoryAdjust,
InventoryCreate,
@@ -378,3 +382,52 @@ async def import_inventory(
unmatched_gtins=[UnmatchedGtin(**g) for g in result.unmatched_gtins],
errors=result.errors,
)
# ============================================================================
# Transaction History Endpoints
# ============================================================================
@router.get("/transactions", response_model=AdminInventoryTransactionListResponse)
def get_all_transactions(
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=200),
vendor_id: int | None = Query(None, description="Filter by vendor"),
product_id: int | None = Query(None, description="Filter by product"),
transaction_type: str | None = Query(None, description="Filter by type"),
order_id: int | None = Query(None, description="Filter by order"),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
):
"""
Get inventory transaction history across all vendors.
Returns a paginated list of all stock movements with vendor and product details.
"""
transactions, total = inventory_transaction_service.get_all_transactions_admin(
db=db,
skip=skip,
limit=limit,
vendor_id=vendor_id,
product_id=product_id,
transaction_type=transaction_type,
order_id=order_id,
)
return AdminInventoryTransactionListResponse(
transactions=[AdminInventoryTransactionItem(**tx) for tx in transactions],
total=total,
skip=skip,
limit=limit,
)
@router.get("/transactions/stats", response_model=AdminTransactionStatsResponse)
def get_transaction_stats(
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
):
"""Get transaction statistics for the platform."""
stats = inventory_transaction_service.get_transaction_stats_admin(db)
return AdminTransactionStatsResponse(**stats)

View File

@@ -289,5 +289,143 @@ class InventoryTransactionService:
}
# =========================================================================
# Admin Methods (cross-vendor operations)
# =========================================================================
def get_all_transactions_admin(
self,
db: Session,
skip: int = 0,
limit: int = 50,
vendor_id: int | None = None,
product_id: int | None = None,
transaction_type: str | None = None,
order_id: int | None = None,
) -> tuple[list[dict], int]:
"""
Get inventory transactions across all vendors (admin only).
Args:
db: Database session
skip: Pagination offset
limit: Pagination limit
vendor_id: Optional vendor filter
product_id: Optional product filter
transaction_type: Optional transaction type filter
order_id: Optional order filter
Returns:
Tuple of (transactions with details, total count)
"""
from models.database.vendor import Vendor
# Build query
query = db.query(InventoryTransaction)
# Apply filters
if vendor_id:
query = query.filter(InventoryTransaction.vendor_id == vendor_id)
if product_id:
query = query.filter(InventoryTransaction.product_id == product_id)
if transaction_type:
query = query.filter(
InventoryTransaction.transaction_type == transaction_type
)
if order_id:
query = query.filter(InventoryTransaction.order_id == order_id)
# Get total count
total = query.count()
# Get transactions with pagination (newest first)
transactions = (
query.order_by(InventoryTransaction.created_at.desc())
.offset(skip)
.limit(limit)
.all()
)
# Build result with vendor and product details
result = []
for tx in transactions:
vendor = db.query(Vendor).filter(Vendor.id == tx.vendor_id).first()
product = db.query(Product).filter(Product.id == tx.product_id).first()
product_title = None
product_sku = None
if product:
product_sku = product.vendor_sku
if product.marketplace_product:
product_title = product.marketplace_product.get_title()
result.append(
{
"id": tx.id,
"vendor_id": tx.vendor_id,
"vendor_name": vendor.name if vendor else None,
"vendor_code": vendor.vendor_code if vendor else None,
"product_id": tx.product_id,
"inventory_id": tx.inventory_id,
"transaction_type": (
tx.transaction_type.value if tx.transaction_type else None
),
"quantity_change": tx.quantity_change,
"quantity_after": tx.quantity_after,
"reserved_after": tx.reserved_after,
"location": tx.location,
"warehouse": tx.warehouse,
"order_id": tx.order_id,
"order_number": tx.order_number,
"reason": tx.reason,
"created_by": tx.created_by,
"created_at": tx.created_at,
"product_title": product_title,
"product_sku": product_sku,
}
)
return result, total
def get_transaction_stats_admin(self, db: Session) -> dict:
"""
Get transaction statistics across the platform (admin only).
Returns:
Dict with transaction counts by type
"""
from sqlalchemy import func as sql_func
# Count by transaction type
type_counts = (
db.query(
InventoryTransaction.transaction_type,
sql_func.count(InventoryTransaction.id).label("count"),
)
.group_by(InventoryTransaction.transaction_type)
.all()
)
# Total transactions
total = db.query(sql_func.count(InventoryTransaction.id)).scalar() or 0
# Transactions today
from datetime import UTC, datetime, timedelta
today_start = datetime.now(UTC).replace(hour=0, minute=0, second=0, microsecond=0)
today_count = (
db.query(sql_func.count(InventoryTransaction.id))
.filter(InventoryTransaction.created_at >= today_start)
.scalar()
or 0
)
return {
"total_transactions": total,
"transactions_today": today_count,
"by_type": {tc.transaction_type.value: tc.count for tc in type_counts},
}
# Create service instance
inventory_transaction_service = InventoryTransactionService()

View File

@@ -263,3 +263,32 @@ class OrderTransactionHistoryResponse(BaseModel):
order_id: int
order_number: str
transactions: list[InventoryTransactionWithProduct]
# ============================================================================
# Admin Inventory Transaction Schemas
# ============================================================================
class AdminInventoryTransactionItem(InventoryTransactionWithProduct):
"""Transaction with vendor details for admin views."""
vendor_name: str | None = None
vendor_code: str | None = None
class AdminInventoryTransactionListResponse(BaseModel):
"""Paginated list of transactions for admin."""
transactions: list[AdminInventoryTransactionItem]
total: int
skip: int
limit: int
class AdminTransactionStatsResponse(BaseModel):
"""Transaction statistics for admin dashboard."""
total_transactions: int
transactions_today: int
by_type: dict[str, int]