249 lines
8.1 KiB
Python
249 lines
8.1 KiB
Python
"""
|
|
Pydantic schemas for POS transaction API requests and responses
|
|
"""
|
|
|
|
from typing import Optional, List, Dict, Any
|
|
from datetime import datetime
|
|
from decimal import Decimal
|
|
from pydantic import BaseModel, Field
|
|
from enum import Enum
|
|
|
|
|
|
class TransactionType(str, Enum):
|
|
"""Transaction type enumeration"""
|
|
SALE = "sale"
|
|
REFUND = "refund"
|
|
VOID = "void"
|
|
EXCHANGE = "exchange"
|
|
|
|
|
|
class TransactionStatus(str, Enum):
|
|
"""Transaction status enumeration"""
|
|
COMPLETED = "completed"
|
|
PENDING = "pending"
|
|
FAILED = "failed"
|
|
REFUNDED = "refunded"
|
|
VOIDED = "voided"
|
|
|
|
|
|
class PaymentMethod(str, Enum):
|
|
"""Payment method enumeration"""
|
|
CARD = "card"
|
|
CASH = "cash"
|
|
DIGITAL_WALLET = "digital_wallet"
|
|
OTHER = "other"
|
|
|
|
|
|
class OrderType(str, Enum):
|
|
"""Order type enumeration"""
|
|
DINE_IN = "dine_in"
|
|
TAKEOUT = "takeout"
|
|
DELIVERY = "delivery"
|
|
PICKUP = "pickup"
|
|
|
|
|
|
class POSTransactionItemResponse(BaseModel):
|
|
"""Schema for POS transaction item response"""
|
|
id: str
|
|
transaction_id: str
|
|
tenant_id: str
|
|
external_item_id: Optional[str] = None
|
|
sku: Optional[str] = None
|
|
product_name: str
|
|
product_category: Optional[str] = None
|
|
product_subcategory: Optional[str] = None
|
|
quantity: Decimal
|
|
unit_price: Decimal
|
|
total_price: Decimal
|
|
discount_amount: Decimal = Decimal("0")
|
|
tax_amount: Decimal = Decimal("0")
|
|
modifiers: Optional[Dict[str, Any]] = None
|
|
inventory_product_id: Optional[str] = None
|
|
is_mapped_to_inventory: bool = False
|
|
is_synced_to_sales: bool = False
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
use_enum_values = True
|
|
json_encoders = {
|
|
datetime: lambda v: v.isoformat() if v else None,
|
|
Decimal: lambda v: float(v) if v else 0.0
|
|
}
|
|
|
|
@classmethod
|
|
def from_orm(cls, obj):
|
|
"""Convert ORM object to schema with proper UUID and Decimal handling"""
|
|
return cls(
|
|
id=str(obj.id),
|
|
transaction_id=str(obj.transaction_id),
|
|
tenant_id=str(obj.tenant_id),
|
|
external_item_id=obj.external_item_id,
|
|
sku=obj.sku,
|
|
product_name=obj.product_name,
|
|
product_category=obj.product_category,
|
|
product_subcategory=obj.product_subcategory,
|
|
quantity=obj.quantity,
|
|
unit_price=obj.unit_price,
|
|
total_price=obj.total_price,
|
|
discount_amount=obj.discount_amount,
|
|
tax_amount=obj.tax_amount,
|
|
modifiers=obj.modifiers,
|
|
inventory_product_id=str(obj.inventory_product_id) if obj.inventory_product_id else None,
|
|
is_mapped_to_inventory=obj.is_mapped_to_inventory,
|
|
is_synced_to_sales=obj.is_synced_to_sales,
|
|
created_at=obj.created_at,
|
|
updated_at=obj.updated_at
|
|
)
|
|
|
|
|
|
class POSTransactionResponse(BaseModel):
|
|
"""Schema for POS transaction response"""
|
|
id: str
|
|
tenant_id: str
|
|
pos_config_id: str
|
|
pos_system: str
|
|
external_transaction_id: str
|
|
external_order_id: Optional[str] = None
|
|
transaction_type: TransactionType
|
|
status: TransactionStatus
|
|
subtotal: Decimal
|
|
tax_amount: Decimal
|
|
tip_amount: Decimal
|
|
discount_amount: Decimal
|
|
total_amount: Decimal
|
|
currency: str = "EUR"
|
|
payment_method: Optional[PaymentMethod] = None
|
|
payment_status: Optional[str] = None
|
|
transaction_date: datetime
|
|
pos_created_at: datetime
|
|
pos_updated_at: Optional[datetime] = None
|
|
location_id: Optional[str] = None
|
|
location_name: Optional[str] = None
|
|
staff_id: Optional[str] = None
|
|
staff_name: Optional[str] = None
|
|
customer_id: Optional[str] = None
|
|
customer_email: Optional[str] = None
|
|
customer_phone: Optional[str] = None
|
|
order_type: Optional[OrderType] = None
|
|
table_number: Optional[str] = None
|
|
receipt_number: Optional[str] = None
|
|
is_synced_to_sales: bool = False
|
|
sales_record_id: Optional[str] = None
|
|
sync_attempted_at: Optional[datetime] = None
|
|
sync_completed_at: Optional[datetime] = None
|
|
sync_error: Optional[str] = None
|
|
sync_retry_count: int = 0
|
|
is_processed: bool = False
|
|
is_duplicate: bool = False
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
items: List[POSTransactionItemResponse] = []
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
use_enum_values = True
|
|
json_encoders = {
|
|
datetime: lambda v: v.isoformat() if v else None,
|
|
Decimal: lambda v: float(v) if v else 0.0
|
|
}
|
|
|
|
@classmethod
|
|
def from_orm(cls, obj):
|
|
"""Convert ORM object to schema with proper UUID and Decimal handling"""
|
|
return cls(
|
|
id=str(obj.id),
|
|
tenant_id=str(obj.tenant_id),
|
|
pos_config_id=str(obj.pos_config_id),
|
|
pos_system=obj.pos_system,
|
|
external_transaction_id=obj.external_transaction_id,
|
|
external_order_id=obj.external_order_id,
|
|
transaction_type=obj.transaction_type,
|
|
status=obj.status,
|
|
subtotal=obj.subtotal,
|
|
tax_amount=obj.tax_amount,
|
|
tip_amount=obj.tip_amount,
|
|
discount_amount=obj.discount_amount,
|
|
total_amount=obj.total_amount,
|
|
currency=obj.currency,
|
|
payment_method=obj.payment_method,
|
|
payment_status=obj.payment_status,
|
|
transaction_date=obj.transaction_date,
|
|
pos_created_at=obj.pos_created_at,
|
|
pos_updated_at=obj.pos_updated_at,
|
|
location_id=obj.location_id,
|
|
location_name=obj.location_name,
|
|
staff_id=obj.staff_id,
|
|
staff_name=obj.staff_name,
|
|
customer_id=obj.customer_id,
|
|
customer_email=obj.customer_email,
|
|
customer_phone=obj.customer_phone,
|
|
order_type=obj.order_type,
|
|
table_number=obj.table_number,
|
|
receipt_number=obj.receipt_number,
|
|
is_synced_to_sales=obj.is_synced_to_sales,
|
|
sales_record_id=str(obj.sales_record_id) if obj.sales_record_id else None,
|
|
sync_attempted_at=obj.sync_attempted_at,
|
|
sync_completed_at=obj.sync_completed_at,
|
|
sync_error=obj.sync_error,
|
|
sync_retry_count=obj.sync_retry_count,
|
|
is_processed=obj.is_processed,
|
|
is_duplicate=obj.is_duplicate,
|
|
created_at=obj.created_at,
|
|
updated_at=obj.updated_at,
|
|
items=[POSTransactionItemResponse.from_orm(item) for item in obj.items] if hasattr(obj, 'items') and obj.items else []
|
|
)
|
|
|
|
|
|
class POSTransactionSummary(BaseModel):
|
|
"""Summary information for a transaction (lightweight)"""
|
|
id: str
|
|
external_transaction_id: str
|
|
transaction_date: datetime
|
|
total_amount: Decimal
|
|
status: TransactionStatus
|
|
payment_method: Optional[PaymentMethod] = None
|
|
is_synced_to_sales: bool
|
|
item_count: int = 0
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
use_enum_values = True
|
|
json_encoders = {
|
|
datetime: lambda v: v.isoformat() if v else None,
|
|
Decimal: lambda v: float(v) if v else 0.0
|
|
}
|
|
|
|
|
|
class POSTransactionListResponse(BaseModel):
|
|
"""Schema for paginated transaction list response"""
|
|
transactions: List[POSTransactionResponse]
|
|
total: int
|
|
has_more: bool = False
|
|
summary: Optional[Dict[str, Any]] = None
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class POSTransactionDashboardSummary(BaseModel):
|
|
"""Dashboard summary for POS transactions"""
|
|
total_transactions_today: int = 0
|
|
total_transactions_this_week: int = 0
|
|
total_transactions_this_month: int = 0
|
|
revenue_today: Decimal = Decimal("0")
|
|
revenue_this_week: Decimal = Decimal("0")
|
|
revenue_this_month: Decimal = Decimal("0")
|
|
average_transaction_value: Decimal = Decimal("0")
|
|
status_breakdown: Dict[str, int] = {}
|
|
payment_method_breakdown: Dict[str, int] = {}
|
|
sync_status: Dict[str, Any] = {}
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
json_encoders = {
|
|
Decimal: lambda v: float(v) if v else 0.0,
|
|
datetime: lambda v: v.isoformat() if v else None
|
|
}
|