From 159243066c21d3317698b650accf0932c426f773 Mon Sep 17 00:00:00 2001 From: Samir Boulahtit Date: Thu, 1 Jan 2026 18:20:12 +0100 Subject: [PATCH] feat: add vendor API endpoints for inventory transaction history MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds three new endpoints for viewing stock movement history: - GET /inventory/transactions - paginated list with filters - GET /inventory/transactions/product/{id} - product-specific history - GET /inventory/transactions/order/{id} - order-specific history Creates InventoryTransactionService to encapsulate query logic following architecture patterns. Includes response schemas with product details for rich UI display. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 --- app/api/v1/vendor/inventory.py | 96 ++++++ app/services/inventory_transaction_service.py | 293 ++++++++++++++++++ models/schema/inventory.py | 63 ++++ 3 files changed, 452 insertions(+) create mode 100644 app/services/inventory_transaction_service.py diff --git a/app/api/v1/vendor/inventory.py b/app/api/v1/vendor/inventory.py index 42db57fc..69266d6f 100644 --- a/app/api/v1/vendor/inventory.py +++ b/app/api/v1/vendor/inventory.py @@ -14,6 +14,7 @@ from sqlalchemy.orm import Session from app.api.deps import get_current_vendor_api from app.core.database import get_db 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 ( InventoryAdjust, @@ -22,8 +23,12 @@ from models.schema.inventory import ( InventoryMessageResponse, InventoryReserve, InventoryResponse, + InventoryTransactionListResponse, + InventoryTransactionWithProduct, InventoryUpdate, + OrderTransactionHistoryResponse, ProductInventorySummary, + ProductTransactionHistoryResponse, ) router = APIRouter() @@ -159,3 +164,94 @@ def delete_inventory( inventory_service.delete_inventory(db, current_user.token_vendor_id, inventory_id) db.commit() return InventoryMessageResponse(message="Inventory deleted successfully") + + +# ============================================================================ +# Inventory Transaction History Endpoints +# ============================================================================ + + +@router.get("/inventory/transactions", response_model=InventoryTransactionListResponse) +def get_inventory_transactions( + skip: int = Query(0, ge=0), + limit: int = Query(50, ge=1, le=200), + product_id: int | None = Query(None, description="Filter by product"), + transaction_type: str | None = Query(None, description="Filter by type"), + current_user: User = Depends(get_current_vendor_api), + db: Session = Depends(get_db), +): + """ + Get inventory transaction history for the vendor. + + Returns a paginated list of all stock movements with product details. + Use filters to narrow down by product or transaction type. + """ + transactions, total = inventory_transaction_service.get_vendor_transactions( + db=db, + vendor_id=current_user.token_vendor_id, + skip=skip, + limit=limit, + product_id=product_id, + transaction_type=transaction_type, + ) + + return InventoryTransactionListResponse( + transactions=[InventoryTransactionWithProduct(**tx) for tx in transactions], + total=total, + skip=skip, + limit=limit, + ) + + +@router.get( + "/inventory/transactions/product/{product_id}", + response_model=ProductTransactionHistoryResponse, +) +def get_product_transaction_history( + product_id: int, + limit: int = Query(50, ge=1, le=200), + current_user: User = Depends(get_current_vendor_api), + db: Session = Depends(get_db), +): + """ + Get transaction history for a specific product. + + Returns recent stock movements with current inventory status. + """ + result = inventory_transaction_service.get_product_history( + db=db, + vendor_id=current_user.token_vendor_id, + product_id=product_id, + limit=limit, + ) + + return ProductTransactionHistoryResponse(**result) + + +@router.get( + "/inventory/transactions/order/{order_id}", + response_model=OrderTransactionHistoryResponse, +) +def get_order_transaction_history( + order_id: int, + current_user: User = Depends(get_current_vendor_api), + db: Session = Depends(get_db), +): + """ + Get all inventory transactions for a specific order. + + Shows all stock movements (reserve, fulfill, release) related to an order. + """ + result = inventory_transaction_service.get_order_history( + db=db, + vendor_id=current_user.token_vendor_id, + order_id=order_id, + ) + + return OrderTransactionHistoryResponse( + order_id=result["order_id"], + order_number=result["order_number"], + transactions=[ + InventoryTransactionWithProduct(**tx) for tx in result["transactions"] + ], + ) diff --git a/app/services/inventory_transaction_service.py b/app/services/inventory_transaction_service.py new file mode 100644 index 00000000..457e5394 --- /dev/null +++ b/app/services/inventory_transaction_service.py @@ -0,0 +1,293 @@ +# app/services/inventory_transaction_service.py +""" +Inventory Transaction Service. + +Provides query operations for inventory transaction history. +All transaction WRITES are handled by OrderInventoryService. +This service handles transaction READS for reporting and auditing. +""" + +import logging +from sqlalchemy import func +from sqlalchemy.orm import Session + +from app.exceptions import OrderNotFoundException, ProductNotFoundException +from models.database.inventory import Inventory +from models.database.inventory_transaction import InventoryTransaction +from models.database.order import Order +from models.database.product import Product + +logger = logging.getLogger(__name__) + + +class InventoryTransactionService: + """Service for querying inventory transaction history.""" + + def get_vendor_transactions( + self, + db: Session, + vendor_id: int, + skip: int = 0, + limit: int = 50, + product_id: int | None = None, + transaction_type: str | None = None, + ) -> tuple[list[dict], int]: + """ + Get inventory transactions for a vendor with optional filters. + + Args: + db: Database session + vendor_id: Vendor ID + skip: Pagination offset + limit: Pagination limit + product_id: Optional product filter + transaction_type: Optional transaction type filter + + Returns: + Tuple of (transactions with product details, total count) + """ + # Build query + query = db.query(InventoryTransaction).filter( + InventoryTransaction.vendor_id == vendor_id + ) + + # Apply filters + if product_id: + query = query.filter(InventoryTransaction.product_id == product_id) + if transaction_type: + query = query.filter( + InventoryTransaction.transaction_type == transaction_type + ) + + # 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 product details + result = [] + for tx in transactions: + 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, + "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_product_history( + self, + db: Session, + vendor_id: int, + product_id: int, + limit: int = 50, + ) -> dict: + """ + Get transaction history for a specific product. + + Args: + db: Database session + vendor_id: Vendor ID + product_id: Product ID + limit: Max transactions to return + + Returns: + Dict with product info, current inventory, and transactions + + Raises: + ProductNotFoundException: If product not found or doesn't belong to vendor + """ + # Get product details + product = ( + db.query(Product) + .filter(Product.id == product_id, Product.vendor_id == vendor_id) + .first() + ) + + if not product: + raise ProductNotFoundException( + f"Product {product_id} not found in vendor catalog" + ) + + product_title = None + product_sku = product.vendor_sku + if product.marketplace_product: + product_title = product.marketplace_product.get_title() + + # Get current inventory + inventory = ( + db.query(Inventory) + .filter(Inventory.product_id == product_id, Inventory.vendor_id == vendor_id) + .first() + ) + + current_quantity = inventory.quantity if inventory else 0 + current_reserved = inventory.reserved_quantity if inventory else 0 + + # Get transactions + transactions = ( + db.query(InventoryTransaction) + .filter( + InventoryTransaction.vendor_id == vendor_id, + InventoryTransaction.product_id == product_id, + ) + .order_by(InventoryTransaction.created_at.desc()) + .limit(limit) + .all() + ) + + total = ( + db.query(func.count(InventoryTransaction.id)) + .filter( + InventoryTransaction.vendor_id == vendor_id, + InventoryTransaction.product_id == product_id, + ) + .scalar() + or 0 + ) + + return { + "product_id": product_id, + "product_title": product_title, + "product_sku": product_sku, + "current_quantity": current_quantity, + "current_reserved": current_reserved, + "transactions": [ + { + "id": tx.id, + "vendor_id": tx.vendor_id, + "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, + } + for tx in transactions + ], + "total": total, + } + + def get_order_history( + self, + db: Session, + vendor_id: int, + order_id: int, + ) -> dict: + """ + Get all inventory transactions for a specific order. + + Args: + db: Database session + vendor_id: Vendor ID + order_id: Order ID + + Returns: + Dict with order info and transactions + + Raises: + OrderNotFoundException: If order not found or doesn't belong to vendor + """ + # Verify order belongs to vendor + order = ( + db.query(Order) + .filter(Order.id == order_id, Order.vendor_id == vendor_id) + .first() + ) + + if not order: + raise OrderNotFoundException(f"Order {order_id} not found") + + # Get transactions for this order + transactions = ( + db.query(InventoryTransaction) + .filter(InventoryTransaction.order_id == order_id) + .order_by(InventoryTransaction.created_at.asc()) + .all() + ) + + # Build result with product details + result = [] + for tx in transactions: + 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, + "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 { + "order_id": order_id, + "order_number": order.order_number, + "transactions": result, + } + + +# Create service instance +inventory_transaction_service = InventoryTransactionService() diff --git a/models/schema/inventory.py b/models/schema/inventory.py index 9a673c05..18ae78d9 100644 --- a/models/schema/inventory.py +++ b/models/schema/inventory.py @@ -200,3 +200,66 @@ class AdminInventoryLocationsResponse(BaseModel): """Response for unique inventory locations.""" locations: list[str] + + +# ============================================================================ +# Inventory Transaction Schemas +# ============================================================================ + + +class InventoryTransactionResponse(BaseModel): + """Single inventory transaction record.""" + + model_config = ConfigDict(from_attributes=True) + + id: int + vendor_id: int + product_id: int + inventory_id: int | None = None + transaction_type: str + quantity_change: int + quantity_after: int + reserved_after: int + location: str | None = None + warehouse: str | None = None + order_id: int | None = None + order_number: str | None = None + reason: str | None = None + created_by: str | None = None + created_at: datetime + + +class InventoryTransactionWithProduct(InventoryTransactionResponse): + """Transaction with product details for list views.""" + + product_title: str | None = None + product_sku: str | None = None + + +class InventoryTransactionListResponse(BaseModel): + """Paginated list of inventory transactions.""" + + transactions: list[InventoryTransactionWithProduct] + total: int + skip: int + limit: int + + +class ProductTransactionHistoryResponse(BaseModel): + """Transaction history for a specific product.""" + + product_id: int + product_title: str | None = None + product_sku: str | None = None + current_quantity: int + current_reserved: int + transactions: list[InventoryTransactionResponse] + total: int + + +class OrderTransactionHistoryResponse(BaseModel): + """Transaction history for a specific order.""" + + order_id: int + order_number: str + transactions: list[InventoryTransactionWithProduct]