390 lines
15 KiB
Python
390 lines
15 KiB
Python
# services/inventory/app/schemas/inventory.py
|
|
"""
|
|
Pydantic schemas for inventory API requests and responses
|
|
"""
|
|
|
|
from typing import Optional, List, Dict, Any
|
|
from datetime import datetime
|
|
from decimal import Decimal
|
|
from pydantic import BaseModel, Field, validator
|
|
from typing import Generic, TypeVar
|
|
from enum import Enum
|
|
|
|
from app.models.inventory import UnitOfMeasure, IngredientCategory, StockMovementType
|
|
|
|
T = TypeVar('T')
|
|
|
|
|
|
# ===== BASE SCHEMAS =====
|
|
|
|
class InventoryBaseSchema(BaseModel):
|
|
"""Base schema for inventory models"""
|
|
|
|
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 None
|
|
}
|
|
|
|
|
|
# ===== INGREDIENT SCHEMAS =====
|
|
|
|
class IngredientCreate(InventoryBaseSchema):
|
|
"""Schema for creating ingredients"""
|
|
name: str = Field(..., max_length=255, description="Ingredient name")
|
|
sku: Optional[str] = Field(None, max_length=100, description="SKU code")
|
|
barcode: Optional[str] = Field(None, max_length=50, description="Barcode")
|
|
category: IngredientCategory = Field(..., description="Ingredient category")
|
|
subcategory: Optional[str] = Field(None, max_length=100, description="Subcategory")
|
|
description: Optional[str] = Field(None, description="Ingredient description")
|
|
brand: Optional[str] = Field(None, max_length=100, description="Brand name")
|
|
unit_of_measure: UnitOfMeasure = Field(..., description="Unit of measure")
|
|
package_size: Optional[float] = Field(None, gt=0, description="Package size")
|
|
|
|
# Pricing
|
|
average_cost: Optional[Decimal] = Field(None, ge=0, description="Average cost per unit")
|
|
standard_cost: Optional[Decimal] = Field(None, ge=0, description="Standard cost per unit")
|
|
|
|
# Stock management
|
|
low_stock_threshold: float = Field(10.0, ge=0, description="Low stock alert threshold")
|
|
reorder_point: float = Field(20.0, ge=0, description="Reorder point")
|
|
reorder_quantity: float = Field(50.0, gt=0, description="Default reorder quantity")
|
|
max_stock_level: Optional[float] = Field(None, gt=0, description="Maximum stock level")
|
|
|
|
# Storage requirements
|
|
requires_refrigeration: bool = Field(False, description="Requires refrigeration")
|
|
requires_freezing: bool = Field(False, description="Requires freezing")
|
|
storage_temperature_min: Optional[float] = Field(None, description="Min storage temperature (°C)")
|
|
storage_temperature_max: Optional[float] = Field(None, description="Max storage temperature (°C)")
|
|
storage_humidity_max: Optional[float] = Field(None, ge=0, le=100, description="Max humidity (%)")
|
|
|
|
# Shelf life
|
|
shelf_life_days: Optional[int] = Field(None, gt=0, description="Shelf life in days")
|
|
storage_instructions: Optional[str] = Field(None, description="Storage instructions")
|
|
|
|
# Properties
|
|
is_perishable: bool = Field(False, description="Is perishable")
|
|
allergen_info: Optional[Dict[str, Any]] = Field(None, description="Allergen information")
|
|
|
|
@validator('storage_temperature_max')
|
|
def validate_temperature_range(cls, v, values):
|
|
if v is not None and 'storage_temperature_min' in values and values['storage_temperature_min'] is not None:
|
|
if v <= values['storage_temperature_min']:
|
|
raise ValueError('Max temperature must be greater than min temperature')
|
|
return v
|
|
|
|
@validator('reorder_point')
|
|
def validate_reorder_point(cls, v, values):
|
|
if 'low_stock_threshold' in values and v <= values['low_stock_threshold']:
|
|
raise ValueError('Reorder point must be greater than low stock threshold')
|
|
return v
|
|
|
|
|
|
class IngredientUpdate(InventoryBaseSchema):
|
|
"""Schema for updating ingredients"""
|
|
name: Optional[str] = Field(None, max_length=255, description="Ingredient name")
|
|
sku: Optional[str] = Field(None, max_length=100, description="SKU code")
|
|
barcode: Optional[str] = Field(None, max_length=50, description="Barcode")
|
|
category: Optional[IngredientCategory] = Field(None, description="Ingredient category")
|
|
subcategory: Optional[str] = Field(None, max_length=100, description="Subcategory")
|
|
description: Optional[str] = Field(None, description="Ingredient description")
|
|
brand: Optional[str] = Field(None, max_length=100, description="Brand name")
|
|
unit_of_measure: Optional[UnitOfMeasure] = Field(None, description="Unit of measure")
|
|
package_size: Optional[float] = Field(None, gt=0, description="Package size")
|
|
|
|
# Pricing
|
|
average_cost: Optional[Decimal] = Field(None, ge=0, description="Average cost per unit")
|
|
standard_cost: Optional[Decimal] = Field(None, ge=0, description="Standard cost per unit")
|
|
|
|
# Stock management
|
|
low_stock_threshold: Optional[float] = Field(None, ge=0, description="Low stock alert threshold")
|
|
reorder_point: Optional[float] = Field(None, ge=0, description="Reorder point")
|
|
reorder_quantity: Optional[float] = Field(None, gt=0, description="Default reorder quantity")
|
|
max_stock_level: Optional[float] = Field(None, gt=0, description="Maximum stock level")
|
|
|
|
# Storage requirements
|
|
requires_refrigeration: Optional[bool] = Field(None, description="Requires refrigeration")
|
|
requires_freezing: Optional[bool] = Field(None, description="Requires freezing")
|
|
storage_temperature_min: Optional[float] = Field(None, description="Min storage temperature (°C)")
|
|
storage_temperature_max: Optional[float] = Field(None, description="Max storage temperature (°C)")
|
|
storage_humidity_max: Optional[float] = Field(None, ge=0, le=100, description="Max humidity (%)")
|
|
|
|
# Shelf life
|
|
shelf_life_days: Optional[int] = Field(None, gt=0, description="Shelf life in days")
|
|
storage_instructions: Optional[str] = Field(None, description="Storage instructions")
|
|
|
|
# Properties
|
|
is_active: Optional[bool] = Field(None, description="Is active")
|
|
is_perishable: Optional[bool] = Field(None, description="Is perishable")
|
|
allergen_info: Optional[Dict[str, Any]] = Field(None, description="Allergen information")
|
|
|
|
|
|
class IngredientResponse(InventoryBaseSchema):
|
|
"""Schema for ingredient API responses"""
|
|
id: str
|
|
tenant_id: str
|
|
name: str
|
|
sku: Optional[str]
|
|
barcode: Optional[str]
|
|
category: IngredientCategory
|
|
subcategory: Optional[str]
|
|
description: Optional[str]
|
|
brand: Optional[str]
|
|
unit_of_measure: UnitOfMeasure
|
|
package_size: Optional[float]
|
|
average_cost: Optional[float]
|
|
last_purchase_price: Optional[float]
|
|
standard_cost: Optional[float]
|
|
low_stock_threshold: float
|
|
reorder_point: float
|
|
reorder_quantity: float
|
|
max_stock_level: Optional[float]
|
|
requires_refrigeration: bool
|
|
requires_freezing: bool
|
|
storage_temperature_min: Optional[float]
|
|
storage_temperature_max: Optional[float]
|
|
storage_humidity_max: Optional[float]
|
|
shelf_life_days: Optional[int]
|
|
storage_instructions: Optional[str]
|
|
is_active: bool
|
|
is_perishable: bool
|
|
allergen_info: Optional[Dict[str, Any]]
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
created_by: Optional[str]
|
|
|
|
# Computed fields
|
|
current_stock: Optional[float] = None
|
|
is_low_stock: Optional[bool] = None
|
|
needs_reorder: Optional[bool] = None
|
|
|
|
|
|
# ===== STOCK SCHEMAS =====
|
|
|
|
class StockCreate(InventoryBaseSchema):
|
|
"""Schema for creating stock entries"""
|
|
ingredient_id: str = Field(..., description="Ingredient ID")
|
|
batch_number: Optional[str] = Field(None, max_length=100, description="Batch number")
|
|
lot_number: Optional[str] = Field(None, max_length=100, description="Lot number")
|
|
supplier_batch_ref: Optional[str] = Field(None, max_length=100, description="Supplier batch reference")
|
|
|
|
current_quantity: float = Field(..., ge=0, description="Current quantity")
|
|
received_date: Optional[datetime] = Field(None, description="Date received")
|
|
expiration_date: Optional[datetime] = Field(None, description="Expiration date")
|
|
best_before_date: Optional[datetime] = Field(None, description="Best before date")
|
|
|
|
unit_cost: Optional[Decimal] = Field(None, ge=0, description="Unit cost")
|
|
storage_location: Optional[str] = Field(None, max_length=100, description="Storage location")
|
|
warehouse_zone: Optional[str] = Field(None, max_length=50, description="Warehouse zone")
|
|
shelf_position: Optional[str] = Field(None, max_length=50, description="Shelf position")
|
|
|
|
quality_status: str = Field("good", description="Quality status")
|
|
|
|
|
|
class StockUpdate(InventoryBaseSchema):
|
|
"""Schema for updating stock entries"""
|
|
batch_number: Optional[str] = Field(None, max_length=100, description="Batch number")
|
|
lot_number: Optional[str] = Field(None, max_length=100, description="Lot number")
|
|
supplier_batch_ref: Optional[str] = Field(None, max_length=100, description="Supplier batch reference")
|
|
|
|
current_quantity: Optional[float] = Field(None, ge=0, description="Current quantity")
|
|
reserved_quantity: Optional[float] = Field(None, ge=0, description="Reserved quantity")
|
|
received_date: Optional[datetime] = Field(None, description="Date received")
|
|
expiration_date: Optional[datetime] = Field(None, description="Expiration date")
|
|
best_before_date: Optional[datetime] = Field(None, description="Best before date")
|
|
|
|
unit_cost: Optional[Decimal] = Field(None, ge=0, description="Unit cost")
|
|
storage_location: Optional[str] = Field(None, max_length=100, description="Storage location")
|
|
warehouse_zone: Optional[str] = Field(None, max_length=50, description="Warehouse zone")
|
|
shelf_position: Optional[str] = Field(None, max_length=50, description="Shelf position")
|
|
|
|
is_available: Optional[bool] = Field(None, description="Is available")
|
|
quality_status: Optional[str] = Field(None, description="Quality status")
|
|
|
|
|
|
class StockResponse(InventoryBaseSchema):
|
|
"""Schema for stock API responses"""
|
|
id: str
|
|
tenant_id: str
|
|
ingredient_id: str
|
|
batch_number: Optional[str]
|
|
lot_number: Optional[str]
|
|
supplier_batch_ref: Optional[str]
|
|
current_quantity: float
|
|
reserved_quantity: float
|
|
available_quantity: float
|
|
received_date: Optional[datetime]
|
|
expiration_date: Optional[datetime]
|
|
best_before_date: Optional[datetime]
|
|
unit_cost: Optional[float]
|
|
total_cost: Optional[float]
|
|
storage_location: Optional[str]
|
|
warehouse_zone: Optional[str]
|
|
shelf_position: Optional[str]
|
|
is_available: bool
|
|
is_expired: bool
|
|
quality_status: str
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
# Related data
|
|
ingredient: Optional[IngredientResponse] = None
|
|
|
|
|
|
# ===== STOCK MOVEMENT SCHEMAS =====
|
|
|
|
class StockMovementCreate(InventoryBaseSchema):
|
|
"""Schema for creating stock movements"""
|
|
ingredient_id: str = Field(..., description="Ingredient ID")
|
|
stock_id: Optional[str] = Field(None, description="Stock ID")
|
|
movement_type: StockMovementType = Field(..., description="Movement type")
|
|
quantity: float = Field(..., description="Quantity moved")
|
|
|
|
unit_cost: Optional[Decimal] = Field(None, ge=0, description="Unit cost")
|
|
reference_number: Optional[str] = Field(None, max_length=100, description="Reference number")
|
|
supplier_id: Optional[str] = Field(None, description="Supplier ID")
|
|
|
|
notes: Optional[str] = Field(None, description="Movement notes")
|
|
reason_code: Optional[str] = Field(None, max_length=50, description="Reason code")
|
|
movement_date: Optional[datetime] = Field(None, description="Movement date")
|
|
|
|
|
|
class StockMovementResponse(InventoryBaseSchema):
|
|
"""Schema for stock movement API responses"""
|
|
id: str
|
|
tenant_id: str
|
|
ingredient_id: str
|
|
stock_id: Optional[str]
|
|
movement_type: StockMovementType
|
|
quantity: float
|
|
unit_cost: Optional[float]
|
|
total_cost: Optional[float]
|
|
quantity_before: Optional[float]
|
|
quantity_after: Optional[float]
|
|
reference_number: Optional[str]
|
|
supplier_id: Optional[str]
|
|
notes: Optional[str]
|
|
reason_code: Optional[str]
|
|
movement_date: datetime
|
|
created_at: datetime
|
|
created_by: Optional[str]
|
|
|
|
# Related data
|
|
ingredient: Optional[IngredientResponse] = None
|
|
|
|
|
|
# ===== ALERT SCHEMAS =====
|
|
|
|
class StockAlertResponse(InventoryBaseSchema):
|
|
"""Schema for stock alert API responses"""
|
|
id: str
|
|
tenant_id: str
|
|
ingredient_id: str
|
|
stock_id: Optional[str]
|
|
alert_type: str
|
|
severity: str
|
|
title: str
|
|
message: str
|
|
current_quantity: Optional[float]
|
|
threshold_value: Optional[float]
|
|
expiration_date: Optional[datetime]
|
|
is_active: bool
|
|
is_acknowledged: bool
|
|
acknowledged_by: Optional[str]
|
|
acknowledged_at: Optional[datetime]
|
|
is_resolved: bool
|
|
resolved_by: Optional[str]
|
|
resolved_at: Optional[datetime]
|
|
resolution_notes: Optional[str]
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
# Related data
|
|
ingredient: Optional[IngredientResponse] = None
|
|
|
|
|
|
# ===== DASHBOARD AND SUMMARY SCHEMAS =====
|
|
|
|
class InventorySummary(InventoryBaseSchema):
|
|
"""Inventory dashboard summary"""
|
|
total_ingredients: int
|
|
total_stock_value: float
|
|
low_stock_alerts: int
|
|
expiring_soon_items: int
|
|
expired_items: int
|
|
out_of_stock_items: int
|
|
|
|
# By category
|
|
stock_by_category: Dict[str, Dict[str, Any]]
|
|
|
|
# Recent activity
|
|
recent_movements: int
|
|
recent_purchases: int
|
|
recent_waste: int
|
|
|
|
|
|
class StockLevelSummary(InventoryBaseSchema):
|
|
"""Stock level summary for an ingredient"""
|
|
ingredient_id: str
|
|
ingredient_name: str
|
|
unit_of_measure: str
|
|
total_quantity: float
|
|
available_quantity: float
|
|
reserved_quantity: float
|
|
|
|
# Status indicators
|
|
is_low_stock: bool
|
|
needs_reorder: bool
|
|
has_expired_stock: bool
|
|
|
|
# Batch information
|
|
total_batches: int
|
|
oldest_batch_date: Optional[datetime]
|
|
newest_batch_date: Optional[datetime]
|
|
next_expiration_date: Optional[datetime]
|
|
|
|
# Cost information
|
|
average_unit_cost: Optional[float]
|
|
total_stock_value: Optional[float]
|
|
|
|
|
|
# ===== REQUEST/RESPONSE WRAPPER SCHEMAS =====
|
|
|
|
class PaginatedResponse(BaseModel, Generic[T]):
|
|
"""Generic paginated response"""
|
|
items: List[T]
|
|
total: int
|
|
page: int
|
|
size: int
|
|
pages: int
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class InventoryFilter(BaseModel):
|
|
"""Inventory filtering parameters"""
|
|
category: Optional[IngredientCategory] = None
|
|
is_active: Optional[bool] = None
|
|
is_low_stock: Optional[bool] = None
|
|
needs_reorder: Optional[bool] = None
|
|
search: Optional[str] = None
|
|
|
|
|
|
class StockFilter(BaseModel):
|
|
"""Stock filtering parameters"""
|
|
ingredient_id: Optional[str] = None
|
|
is_available: Optional[bool] = None
|
|
is_expired: Optional[bool] = None
|
|
expiring_within_days: Optional[int] = None
|
|
storage_location: Optional[str] = None
|
|
quality_status: Optional[str] = None
|
|
|
|
|
|
# Type aliases for paginated responses
|
|
IngredientListResponse = PaginatedResponse[IngredientResponse]
|
|
StockListResponse = PaginatedResponse[StockResponse]
|
|
StockMovementListResponse = PaginatedResponse[StockMovementResponse]
|
|
StockAlertListResponse = PaginatedResponse[StockAlertResponse] |