# app/services/order_service.py """ Unified order service for all sales channels. This service handles: - Order creation (direct and marketplace) - Order status management - Order retrieval and filtering - Customer creation for marketplace imports - Order item management All orders use snapshotted customer and address data. """ import logging import random import string from datetime import UTC, datetime from typing import Any from sqlalchemy import and_, func, or_ from sqlalchemy.orm import Session from app.exceptions import ( CustomerNotFoundException, InsufficientInventoryException, OrderNotFoundException, ValidationException, ) from app.services.order_item_exception_service import order_item_exception_service from models.database.customer import Customer from models.database.marketplace_product import MarketplaceProduct from models.database.marketplace_product_translation import MarketplaceProductTranslation from models.database.order import Order, OrderItem from models.database.product import Product from models.database.vendor import Vendor # Placeholder product constants PLACEHOLDER_GTIN = "0000000000000" PLACEHOLDER_MARKETPLACE_ID = "PLACEHOLDER" from models.schema.order import ( AddressSnapshot, CustomerSnapshot, OrderCreate, OrderItemCreate, OrderUpdate, ) logger = logging.getLogger(__name__) class OrderService: """Unified service for order operations across all channels.""" # ========================================================================= # Order Number Generation # ========================================================================= def _generate_order_number(self, db: Session, vendor_id: int) -> str: """ Generate unique order number. Format: ORD-{VENDOR_ID}-{TIMESTAMP}-{RANDOM} Example: ORD-1-20250110-A1B2C3 """ timestamp = datetime.now(UTC).strftime("%Y%m%d") random_suffix = "".join( random.choices(string.ascii_uppercase + string.digits, k=6) ) order_number = f"ORD-{vendor_id}-{timestamp}-{random_suffix}" # Ensure uniqueness while db.query(Order).filter(Order.order_number == order_number).first(): random_suffix = "".join( random.choices(string.ascii_uppercase + string.digits, k=6) ) order_number = f"ORD-{vendor_id}-{timestamp}-{random_suffix}" return order_number # ========================================================================= # Placeholder Product Management # ========================================================================= def _get_or_create_placeholder_product( self, db: Session, vendor_id: int, ) -> Product: """ Get or create the vendor's placeholder product for unmatched items. When a marketplace order contains a GTIN that doesn't match any product in the vendor's catalog, we link the order item to this placeholder and create an exception for resolution. Args: db: Database session vendor_id: Vendor ID Returns: Placeholder Product for the vendor """ # Check for existing placeholder product for this vendor placeholder = ( db.query(Product) .filter( and_( Product.vendor_id == vendor_id, Product.gtin == PLACEHOLDER_GTIN, ) ) .first() ) if placeholder: return placeholder # Get or create placeholder marketplace product (shared) mp = ( db.query(MarketplaceProduct) .filter( MarketplaceProduct.marketplace_product_id == PLACEHOLDER_MARKETPLACE_ID ) .first() ) if not mp: mp = MarketplaceProduct( marketplace_product_id=PLACEHOLDER_MARKETPLACE_ID, marketplace="internal", vendor_name="system", product_type_enum="physical", is_active=False, # Not for sale ) db.add(mp) db.flush() # Add translation for placeholder translation = MarketplaceProductTranslation( marketplace_product_id=mp.id, language="en", title="Unmatched Product (Pending Resolution)", description=( "This is a placeholder for products not found during order import. " "Please resolve the exception to assign the correct product." ), ) db.add(translation) db.flush() logger.info( f"Created placeholder MarketplaceProduct {mp.id}" ) # Create vendor-specific placeholder product placeholder = Product( vendor_id=vendor_id, marketplace_product_id=mp.id, gtin=PLACEHOLDER_GTIN, gtin_type="placeholder", is_active=False, ) db.add(placeholder) db.flush() logger.info( f"Created placeholder product {placeholder.id} for vendor {vendor_id}" ) return placeholder # ========================================================================= # Customer Management # ========================================================================= def find_or_create_customer( self, db: Session, vendor_id: int, email: str, first_name: str, last_name: str, phone: str | None = None, is_active: bool = False, ) -> Customer: """ Find existing customer by email or create new one. For marketplace imports, customers are created as inactive until they register on the storefront. Args: db: Database session vendor_id: Vendor ID email: Customer email first_name: Customer first name last_name: Customer last name phone: Customer phone (optional) is_active: Whether customer is active (default: False for imports) Returns: Customer record (existing or newly created) """ # Look for existing customer by email within vendor scope customer = ( db.query(Customer) .filter( and_( Customer.vendor_id == vendor_id, Customer.email == email, ) ) .first() ) if customer: return customer # Generate a unique customer number timestamp = datetime.now(UTC).strftime("%Y%m%d%H%M%S") random_suffix = "".join(random.choices(string.digits, k=4)) customer_number = f"CUST-{vendor_id}-{timestamp}-{random_suffix}" # Create new customer customer = Customer( vendor_id=vendor_id, email=email, first_name=first_name, last_name=last_name, phone=phone, customer_number=customer_number, hashed_password="", # No password for imported customers is_active=is_active, ) db.add(customer) db.flush() logger.info( f"Created {'active' if is_active else 'inactive'} customer " f"{customer.id} for vendor {vendor_id}: {email}" ) return customer # ========================================================================= # Order Creation # ========================================================================= def create_order( self, db: Session, vendor_id: int, order_data: OrderCreate, ) -> Order: """ Create a new direct order. Args: db: Database session vendor_id: Vendor ID order_data: Order creation data Returns: Created Order object Raises: ValidationException: If order data is invalid InsufficientInventoryException: If not enough inventory """ try: # Get or create customer if order_data.customer_id: customer = ( db.query(Customer) .filter( and_( Customer.id == order_data.customer_id, Customer.vendor_id == vendor_id, ) ) .first() ) if not customer: raise CustomerNotFoundException(str(order_data.customer_id)) else: # Create customer from snapshot customer = self.find_or_create_customer( db=db, vendor_id=vendor_id, email=order_data.customer.email, first_name=order_data.customer.first_name, last_name=order_data.customer.last_name, phone=order_data.customer.phone, is_active=True, # Direct orders = active customers ) # Calculate order totals and validate products subtotal = 0.0 order_items_data = [] for item_data in order_data.items: product = ( db.query(Product) .filter( and_( Product.id == item_data.product_id, Product.vendor_id == vendor_id, Product.is_active == True, ) ) .first() ) if not product: raise ValidationException( f"Product {item_data.product_id} not found" ) # Check inventory if product.available_inventory < item_data.quantity: raise InsufficientInventoryException( product_id=product.id, requested=item_data.quantity, available=product.available_inventory, ) # Calculate item total unit_price = product.sale_price if product.sale_price else product.price if not unit_price: raise ValidationException(f"Product {product.id} has no price") item_total = unit_price * item_data.quantity subtotal += item_total order_items_data.append( { "product_id": product.id, "product_name": product.marketplace_product.title if product.marketplace_product else product.product_id, "product_sku": product.product_id, "gtin": product.gtin, "gtin_type": product.gtin_type, "quantity": item_data.quantity, "unit_price": unit_price, "total_price": item_total, } ) # Calculate totals tax_amount = 0.0 # TODO: Implement tax calculation shipping_amount = 5.99 if subtotal < 50 else 0.0 discount_amount = 0.0 total_amount = subtotal + tax_amount + shipping_amount - discount_amount # Use billing address or shipping address billing = order_data.billing_address or order_data.shipping_address # Generate order number order_number = self._generate_order_number(db, vendor_id) # Create order with snapshots order = Order( vendor_id=vendor_id, customer_id=customer.id, order_number=order_number, channel="direct", status="pending", # Financials subtotal=subtotal, tax_amount=tax_amount, shipping_amount=shipping_amount, discount_amount=discount_amount, total_amount=total_amount, currency="EUR", # Customer snapshot customer_first_name=order_data.customer.first_name, customer_last_name=order_data.customer.last_name, customer_email=order_data.customer.email, customer_phone=order_data.customer.phone, customer_locale=order_data.customer.locale, # Shipping address snapshot ship_first_name=order_data.shipping_address.first_name, ship_last_name=order_data.shipping_address.last_name, ship_company=order_data.shipping_address.company, ship_address_line_1=order_data.shipping_address.address_line_1, ship_address_line_2=order_data.shipping_address.address_line_2, ship_city=order_data.shipping_address.city, ship_postal_code=order_data.shipping_address.postal_code, ship_country_iso=order_data.shipping_address.country_iso, # Billing address snapshot bill_first_name=billing.first_name, bill_last_name=billing.last_name, bill_company=billing.company, bill_address_line_1=billing.address_line_1, bill_address_line_2=billing.address_line_2, bill_city=billing.city, bill_postal_code=billing.postal_code, bill_country_iso=billing.country_iso, # Other shipping_method=order_data.shipping_method, customer_notes=order_data.customer_notes, order_date=datetime.now(UTC), ) db.add(order) db.flush() # Create order items for item_data in order_items_data: order_item = OrderItem(order_id=order.id, **item_data) db.add(order_item) db.flush() db.refresh(order) logger.info( f"Order {order.order_number} created for vendor {vendor_id}, " f"total: EUR {total_amount:.2f}" ) return order except ( ValidationException, InsufficientInventoryException, CustomerNotFoundException, ): raise except Exception as e: logger.error(f"Error creating order: {str(e)}") raise ValidationException(f"Failed to create order: {str(e)}") def create_letzshop_order( self, db: Session, vendor_id: int, shipment_data: dict[str, Any], ) -> Order: """ Create an order from Letzshop shipment data. Validates all products exist BEFORE creating any database records. This ensures we don't leave the session in an inconsistent state if validation fails. Args: db: Database session vendor_id: Vendor ID shipment_data: Raw shipment data from Letzshop API Returns: Created Order object Raises: ValidationException: If product not found by GTIN """ order_data = shipment_data.get("order", {}) # Generate order number using Letzshop order number letzshop_order_number = order_data.get("number", "") order_number = f"LS-{vendor_id}-{letzshop_order_number}" # Check if order already exists (read-only, safe to do first) existing = ( db.query(Order) .filter(Order.order_number == order_number) .first() ) if existing: return existing # ===================================================================== # PHASE 1: Parse and validate all data BEFORE any database writes # ===================================================================== # Parse inventory units inventory_units = shipment_data.get("inventoryUnits", []) if isinstance(inventory_units, dict): inventory_units = inventory_units.get("nodes", []) # Collect all GTINs and check for items without GTINs gtins = set() has_items_without_gtin = False for unit in inventory_units: variant = unit.get("variant", {}) or {} trade_id = variant.get("tradeId", {}) or {} gtin = trade_id.get("number") if gtin: gtins.add(gtin) else: has_items_without_gtin = True # Batch query all products by GTIN products_by_gtin: dict[str, Product] = {} if gtins: products = ( db.query(Product) .filter( and_( Product.vendor_id == vendor_id, Product.gtin.in_(gtins), ) ) .all() ) products_by_gtin = {p.gtin: p for p in products if p.gtin} # Identify missing GTINs (graceful handling - no exception raised) missing_gtins = gtins - set(products_by_gtin.keys()) placeholder = None if missing_gtins or has_items_without_gtin: # Get or create placeholder product for unmatched items or items without GTIN placeholder = self._get_or_create_placeholder_product(db, vendor_id) if missing_gtins: logger.warning( f"Order {order_number}: {len(missing_gtins)} product(s) not found. " f"GTINs: {missing_gtins}. Using placeholder and creating exceptions." ) if has_items_without_gtin: logger.warning( f"Order {order_number}: Some items have no GTIN. " f"Using placeholder and creating exceptions." ) # Parse address data ship_address = order_data.get("shipAddress", {}) or {} bill_address = order_data.get("billAddress", {}) or {} ship_country = ship_address.get("country", {}) or {} bill_country = bill_address.get("country", {}) or {} # Extract customer info customer_email = order_data.get("email", "") ship_first_name = ship_address.get("firstName", "") or "" ship_last_name = ship_address.get("lastName", "") or "" # Parse order date order_date = datetime.now(UTC) completed_at_str = order_data.get("completedAt") if completed_at_str: try: if completed_at_str.endswith("Z"): completed_at_str = completed_at_str[:-1] + "+00:00" order_date = datetime.fromisoformat(completed_at_str) except (ValueError, TypeError): pass # Parse total amount total_str = order_data.get("total", "0") try: # Handle format like "99.99 EUR" total_amount = float(str(total_str).split()[0]) except (ValueError, IndexError): total_amount = 0.0 # Map Letzshop state to status letzshop_state = shipment_data.get("state", "unconfirmed") status_mapping = { "unconfirmed": "pending", "confirmed": "processing", "declined": "cancelled", } status = status_mapping.get(letzshop_state, "pending") # ===================================================================== # PHASE 2: All validation passed - now create database records # ===================================================================== # Find or create customer (inactive) customer = self.find_or_create_customer( db=db, vendor_id=vendor_id, email=customer_email, first_name=ship_first_name, last_name=ship_last_name, is_active=False, ) # Create order order = Order( vendor_id=vendor_id, customer_id=customer.id, order_number=order_number, channel="letzshop", # External references external_order_id=order_data.get("id"), external_shipment_id=shipment_data.get("id"), external_order_number=letzshop_order_number, external_data=shipment_data, # Status status=status, # Financials total_amount=total_amount, currency="EUR", # Customer snapshot customer_first_name=ship_first_name, customer_last_name=ship_last_name, customer_email=customer_email, customer_locale=order_data.get("locale"), # Shipping address snapshot ship_first_name=ship_first_name, ship_last_name=ship_last_name, ship_company=ship_address.get("company"), ship_address_line_1=ship_address.get("streetName", "") or "", ship_address_line_2=ship_address.get("streetNumber"), ship_city=ship_address.get("city", "") or "", ship_postal_code=ship_address.get("postalCode", "") or "", ship_country_iso=ship_country.get("iso", "") or "", # Billing address snapshot bill_first_name=bill_address.get("firstName", "") or ship_first_name, bill_last_name=bill_address.get("lastName", "") or ship_last_name, bill_company=bill_address.get("company"), bill_address_line_1=bill_address.get("streetName", "") or "", bill_address_line_2=bill_address.get("streetNumber"), bill_city=bill_address.get("city", "") or "", bill_postal_code=bill_address.get("postalCode", "") or "", bill_country_iso=bill_country.get("iso", "") or "", # Dates order_date=order_date, confirmed_at=datetime.now(UTC) if status == "processing" else None, cancelled_at=datetime.now(UTC) if status == "cancelled" else None, ) db.add(order) db.flush() # Create order items from inventory units exceptions_created = 0 for unit in inventory_units: variant = unit.get("variant", {}) or {} product_info = variant.get("product", {}) or {} trade_id = variant.get("tradeId", {}) or {} product_name_dict = product_info.get("name", {}) or {} gtin = trade_id.get("number") gtin_type = trade_id.get("parser") # Get product from map, or use placeholder if not found product = products_by_gtin.get(gtin) needs_product_match = False if not product: # Use placeholder for unmatched items product = placeholder needs_product_match = True # Get product name from marketplace data product_name = ( product_name_dict.get("en") or product_name_dict.get("fr") or product_name_dict.get("de") or str(product_name_dict) ) # Get price unit_price = 0.0 price_str = variant.get("price", "0") try: unit_price = float(str(price_str).split()[0]) except (ValueError, IndexError): pass # Map item state item_state = unit.get("state") # unconfirmed, confirmed_available, etc. order_item = OrderItem( order_id=order.id, product_id=product.id if product else None, product_name=product_name, product_sku=variant.get("sku"), gtin=gtin, gtin_type=gtin_type, quantity=1, # Letzshop uses individual inventory units unit_price=unit_price, total_price=unit_price, external_item_id=unit.get("id"), external_variant_id=variant.get("id"), item_state=item_state, needs_product_match=needs_product_match, ) db.add(order_item) db.flush() # Get order_item.id for exception creation # Create exception record for unmatched items if needs_product_match: order_item_exception_service.create_exception( db=db, order_item=order_item, vendor_id=vendor_id, original_gtin=gtin, original_product_name=product_name, original_sku=variant.get("sku"), exception_type="product_not_found", ) exceptions_created += 1 db.flush() db.refresh(order) if exceptions_created > 0: logger.info( f"Created {exceptions_created} order item exception(s) for " f"order {order.order_number}" ) logger.info( f"Letzshop order {order.order_number} created for vendor {vendor_id}, " f"status: {status}, items: {len(inventory_units)}" ) return order # ========================================================================= # Order Retrieval # ========================================================================= def get_order(self, db: Session, vendor_id: int, order_id: int) -> Order: """Get order by ID within vendor scope.""" order = ( db.query(Order) .filter(and_(Order.id == order_id, Order.vendor_id == vendor_id)) .first() ) if not order: raise OrderNotFoundException(str(order_id)) return order def get_order_by_external_shipment_id( self, db: Session, vendor_id: int, shipment_id: str, ) -> Order | None: """Get order by external shipment ID (for Letzshop).""" return ( db.query(Order) .filter( and_( Order.vendor_id == vendor_id, Order.external_shipment_id == shipment_id, ) ) .first() ) def get_vendor_orders( self, db: Session, vendor_id: int, skip: int = 0, limit: int = 50, status: str | None = None, channel: str | None = None, search: str | None = None, ) -> tuple[list[Order], int]: """ Get orders for vendor with filtering. Args: db: Database session vendor_id: Vendor ID skip: Pagination offset limit: Pagination limit status: Filter by status channel: Filter by channel (direct, letzshop) search: Search by order number, customer name, or email Returns: Tuple of (orders, total_count) """ query = db.query(Order).filter(Order.vendor_id == vendor_id) if status: query = query.filter(Order.status == status) if channel: query = query.filter(Order.channel == channel) if search: search_term = f"%{search}%" query = query.filter( or_( Order.order_number.ilike(search_term), Order.external_order_number.ilike(search_term), Order.customer_email.ilike(search_term), Order.customer_first_name.ilike(search_term), Order.customer_last_name.ilike(search_term), ) ) # Order by most recent first query = query.order_by(Order.order_date.desc()) total = query.count() orders = query.offset(skip).limit(limit).all() return orders, total def get_order_stats(self, db: Session, vendor_id: int) -> dict[str, int]: """ Get order counts by status for a vendor. Returns: Dict with counts for each status. """ status_counts = ( db.query(Order.status, func.count(Order.id).label("count")) .filter(Order.vendor_id == vendor_id) .group_by(Order.status) .all() ) stats = { "pending": 0, "processing": 0, "shipped": 0, "delivered": 0, "cancelled": 0, "refunded": 0, "total": 0, } for status, count in status_counts: if status in stats: stats[status] = count stats["total"] += count # Also count by channel channel_counts = ( db.query(Order.channel, func.count(Order.id).label("count")) .filter(Order.vendor_id == vendor_id) .group_by(Order.channel) .all() ) for channel, count in channel_counts: stats[f"{channel}_orders"] = count return stats # ========================================================================= # Order Updates # ========================================================================= def update_order_status( self, db: Session, vendor_id: int, order_id: int, order_update: OrderUpdate, ) -> Order: """ Update order status and tracking information. Args: db: Database session vendor_id: Vendor ID order_id: Order ID order_update: Update data Returns: Updated Order object """ order = self.get_order(db, vendor_id, order_id) now = datetime.now(UTC) if order_update.status: old_status = order.status order.status = order_update.status # Update timestamps based on status if order_update.status == "processing" and not order.confirmed_at: order.confirmed_at = now elif order_update.status == "shipped" and not order.shipped_at: order.shipped_at = now elif order_update.status == "delivered" and not order.delivered_at: order.delivered_at = now elif order_update.status == "cancelled" and not order.cancelled_at: order.cancelled_at = now if order_update.tracking_number: order.tracking_number = order_update.tracking_number if order_update.tracking_provider: order.tracking_provider = order_update.tracking_provider if order_update.internal_notes: order.internal_notes = order_update.internal_notes order.updated_at = now db.flush() db.refresh(order) logger.info(f"Order {order.order_number} updated: status={order.status}") return order def set_order_tracking( self, db: Session, vendor_id: int, order_id: int, tracking_number: str, tracking_provider: str, ) -> Order: """ Set tracking information and mark as shipped. Args: db: Database session vendor_id: Vendor ID order_id: Order ID tracking_number: Tracking number tracking_provider: Shipping provider Returns: Updated Order object """ order = self.get_order(db, vendor_id, order_id) now = datetime.now(UTC) order.tracking_number = tracking_number order.tracking_provider = tracking_provider order.status = "shipped" order.shipped_at = now order.updated_at = now db.flush() db.refresh(order) logger.info( f"Order {order.order_number} shipped: " f"{tracking_provider} {tracking_number}" ) return order def update_item_state( self, db: Session, vendor_id: int, order_id: int, item_id: int, state: str, ) -> OrderItem: """ Update the state of an order item (for marketplace confirmation). Args: db: Database session vendor_id: Vendor ID order_id: Order ID item_id: Item ID state: New state (confirmed_available, confirmed_unavailable) Returns: Updated OrderItem object """ order = self.get_order(db, vendor_id, order_id) item = ( db.query(OrderItem) .filter( and_( OrderItem.id == item_id, OrderItem.order_id == order.id, ) ) .first() ) if not item: raise ValidationException(f"Order item {item_id} not found") item.item_state = state item.updated_at = datetime.now(UTC) # Check if all items are processed all_items = db.query(OrderItem).filter(OrderItem.order_id == order.id).all() all_confirmed = all( i.item_state in ("confirmed_available", "confirmed_unavailable") for i in all_items ) if all_confirmed: has_available = any( i.item_state == "confirmed_available" for i in all_items ) all_unavailable = all( i.item_state == "confirmed_unavailable" for i in all_items ) now = datetime.now(UTC) if all_unavailable: order.status = "cancelled" order.cancelled_at = now elif has_available: order.status = "processing" order.confirmed_at = now order.updated_at = now db.flush() db.refresh(item) return item # ========================================================================= # Admin Methods (cross-vendor) # ========================================================================= def get_all_orders_admin( self, db: Session, skip: int = 0, limit: int = 50, vendor_id: int | None = None, status: str | None = None, channel: str | None = None, search: str | None = None, ) -> tuple[list[dict], int]: """ Get orders across all vendors for admin. Args: db: Database session skip: Pagination offset limit: Pagination limit vendor_id: Filter by vendor status: Filter by status channel: Filter by channel search: Search by order number or customer Returns: Tuple of (orders with vendor info, total_count) """ query = db.query(Order).join(Vendor) if vendor_id: query = query.filter(Order.vendor_id == vendor_id) if status: query = query.filter(Order.status == status) if channel: query = query.filter(Order.channel == channel) if search: search_term = f"%{search}%" query = query.filter( or_( Order.order_number.ilike(search_term), Order.external_order_number.ilike(search_term), Order.customer_email.ilike(search_term), Order.customer_first_name.ilike(search_term), Order.customer_last_name.ilike(search_term), ) ) query = query.order_by(Order.order_date.desc()) total = query.count() orders = query.offset(skip).limit(limit).all() result = [] for order in orders: item_count = len(order.items) if order.items else 0 result.append( { "id": order.id, "vendor_id": order.vendor_id, "vendor_name": order.vendor.name if order.vendor else None, "vendor_code": order.vendor.vendor_code if order.vendor else None, "customer_id": order.customer_id, "customer_full_name": order.customer_full_name, "customer_email": order.customer_email, "order_number": order.order_number, "channel": order.channel, "status": order.status, "external_order_number": order.external_order_number, "external_shipment_id": order.external_shipment_id, "subtotal": order.subtotal, "tax_amount": order.tax_amount, "shipping_amount": order.shipping_amount, "discount_amount": order.discount_amount, "total_amount": order.total_amount, "currency": order.currency, "ship_country_iso": order.ship_country_iso, "tracking_number": order.tracking_number, "tracking_provider": order.tracking_provider, "item_count": item_count, "order_date": order.order_date, "confirmed_at": order.confirmed_at, "shipped_at": order.shipped_at, "delivered_at": order.delivered_at, "cancelled_at": order.cancelled_at, "created_at": order.created_at, "updated_at": order.updated_at, } ) return result, total def get_order_stats_admin(self, db: Session) -> dict: """Get platform-wide order statistics.""" # Get status counts status_counts = ( db.query(Order.status, func.count(Order.id)) .group_by(Order.status) .all() ) stats = { "total_orders": 0, "pending_orders": 0, "processing_orders": 0, "shipped_orders": 0, "delivered_orders": 0, "cancelled_orders": 0, "refunded_orders": 0, "total_revenue": 0.0, "direct_orders": 0, "letzshop_orders": 0, "vendors_with_orders": 0, } for status, count in status_counts: stats["total_orders"] += count key = f"{status}_orders" if key in stats: stats[key] = count # Get channel counts channel_counts = ( db.query(Order.channel, func.count(Order.id)) .group_by(Order.channel) .all() ) for channel, count in channel_counts: key = f"{channel}_orders" if key in stats: stats[key] = count # Get total revenue (from delivered orders) revenue = ( db.query(func.sum(Order.total_amount)) .filter(Order.status == "delivered") .scalar() ) stats["total_revenue"] = float(revenue) if revenue else 0.0 # Count vendors with orders vendors_count = ( db.query(func.count(func.distinct(Order.vendor_id))).scalar() or 0 ) stats["vendors_with_orders"] = vendors_count return stats def get_order_by_id_admin(self, db: Session, order_id: int) -> Order: """Get order by ID without vendor scope (admin only).""" order = db.query(Order).filter(Order.id == order_id).first() if not order: raise OrderNotFoundException(str(order_id)) return order def get_vendors_with_orders_admin(self, db: Session) -> list[dict]: """Get list of vendors that have orders (admin only).""" from models.database.vendor import Vendor # Query vendors with order counts results = ( db.query( Vendor.id, Vendor.name, Vendor.vendor_code, func.count(Order.id).label("order_count"), ) .join(Order, Order.vendor_id == Vendor.id) .group_by(Vendor.id, Vendor.name, Vendor.vendor_code) .order_by(func.count(Order.id).desc()) .all() ) return [ { "id": row.id, "name": row.name, "vendor_code": row.vendor_code, "order_count": row.order_count, } for row in results ] # Create service instance order_service = OrderService()