sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
82c07c165fdde9c2d0ee453e56823530385bf025
orion/models/schema/order.py
Samir Boulahtit 82c07c165f feat: add customer profile, VAT alignment, and fix shop auth 
Customer Profile:
- Add profile API (GET/PUT /api/v1/shop/profile)
- Add password change endpoint (PUT /api/v1/shop/profile/password)
- Implement full profile page with preferences and password sections
- Add CustomerPasswordChange schema

Shop Authentication Fixes:
- Add Authorization header to all shop account API calls
- Fix orders, order-detail, messages pages authentication
- Add proper redirect to login on 401 responses
- Fix toast message showing noqa comment in shop-layout.js

VAT Calculation:
- Add shared VAT utility (app/utils/vat.py)
- Add VAT fields to Order model (vat_regime, vat_rate, etc.)
- Align order VAT calculation with invoice settings
- Add migration for VAT fields on orders

Validation Framework:
- Fix base_validator.py with missing methods
- Add validate_file, output_results, get_exit_code methods
- Fix validate_all.py import issues

Documentation:
- Add launch-readiness.md tracking OMS status
- Update to 95% feature complete

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-02 20:31:48 +01:00

585 lines
15 KiB
Python
Raw Blame History

# models/schema/order.py
"""
Pydantic schemas for unified order operations.
Supports both direct orders and marketplace orders (Letzshop, etc.)
with snapshotted customer and address data.
"""
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
# ============================================================================
# Address Snapshot Schemas
# ============================================================================
class AddressSnapshot(BaseModel):
"""Address snapshot for order creation."""
first_name: str = Field(..., min_length=1, max_length=100)
last_name: str = Field(..., min_length=1, max_length=100)
company: str | None = Field(None, max_length=200)
address_line_1: str = Field(..., min_length=1, max_length=255)
address_line_2: str | None = Field(None, max_length=255)
city: str = Field(..., min_length=1, max_length=100)
postal_code: str = Field(..., min_length=1, max_length=20)
country_iso: str = Field(..., min_length=2, max_length=5)
class AddressSnapshotResponse(BaseModel):
"""Address snapshot in order response."""
first_name: str
last_name: str
company: str | None
address_line_1: str
address_line_2: str | None
city: str
postal_code: str
country_iso: str
@property
def full_name(self) -> str:
return f"{self.first_name} {self.last_name}".strip()
# ============================================================================
# Order Item Schemas
# ============================================================================
class OrderItemCreate(BaseModel):
"""Schema for creating an order item."""
product_id: int
quantity: int = Field(..., ge=1)
class OrderItemExceptionBrief(BaseModel):
"""Brief exception info for embedding in order item responses."""
model_config = ConfigDict(from_attributes=True)
id: int
original_gtin: str | None
original_product_name: str | None
exception_type: str
status: str
resolved_product_id: int | None
class OrderItemResponse(BaseModel):
"""Schema for order item response."""
model_config = ConfigDict(from_attributes=True)
id: int
order_id: int
product_id: int
product_name: str
product_sku: str | None
gtin: str | None
gtin_type: str | None
quantity: int
unit_price: float
total_price: float
# External references (for marketplace items)
external_item_id: str | None = None
external_variant_id: str | None = None
# Item state (for marketplace confirmation flow)
item_state: str | None = None
# Inventory tracking
inventory_reserved: bool
inventory_fulfilled: bool
# Exception tracking
needs_product_match: bool = False
exception: OrderItemExceptionBrief | None = None
created_at: datetime
updated_at: datetime
@property
def is_confirmed(self) -> bool:
"""Check if item has been confirmed (available or unavailable)."""
return self.item_state in ("confirmed_available", "confirmed_unavailable")
@property
def is_available(self) -> bool:
"""Check if item is confirmed as available."""
return self.item_state == "confirmed_available"
@property
def is_declined(self) -> bool:
"""Check if item was declined (unavailable)."""
return self.item_state == "confirmed_unavailable"
@property
def has_unresolved_exception(self) -> bool:
"""Check if item has an unresolved exception blocking confirmation."""
if not self.exception:
return False
return self.exception.status in ("pending", "ignored")
# ============================================================================
# Customer Snapshot Schemas
# ============================================================================
class CustomerSnapshot(BaseModel):
"""Customer snapshot for order creation."""
first_name: str = Field(..., min_length=1, max_length=100)
last_name: str = Field(..., min_length=1, max_length=100)
email: str = Field(..., max_length=255)
phone: str | None = Field(None, max_length=50)
locale: str | None = Field(None, max_length=10)
class CustomerSnapshotResponse(BaseModel):
"""Customer snapshot in order response."""
first_name: str
last_name: str
email: str
phone: str | None
locale: str | None
@property
def full_name(self) -> str:
return f"{self.first_name} {self.last_name}".strip()
# ============================================================================
# Order Create/Update Schemas
# ============================================================================
class OrderCreate(BaseModel):
"""Schema for creating an order (direct channel)."""
customer_id: int | None = None # Optional for guest checkout
items: list[OrderItemCreate] = Field(..., min_length=1)
# Customer info snapshot
customer: CustomerSnapshot
# Addresses (snapshots)
shipping_address: AddressSnapshot
billing_address: AddressSnapshot | None = None # Use shipping if not provided
# Optional fields
shipping_method: str | None = None
customer_notes: str | None = Field(None, max_length=1000)
# Cart/session info
session_id: str | None = None
class OrderUpdate(BaseModel):
"""Schema for updating order status."""
status: str | None = Field(
None, pattern="^(pending|processing|shipped|delivered|cancelled|refunded)$"
)
tracking_number: str | None = None
tracking_provider: str | None = None
internal_notes: str | None = None
class OrderTrackingUpdate(BaseModel):
"""Schema for setting tracking information."""
tracking_number: str = Field(..., min_length=1, max_length=100)
tracking_provider: str = Field(..., min_length=1, max_length=100)
class OrderItemStateUpdate(BaseModel):
"""Schema for updating item state (marketplace confirmation)."""
item_id: int
state: str = Field(..., pattern="^(confirmed_available|confirmed_unavailable)$")
# ============================================================================
# Order Response Schemas
# ============================================================================
class OrderResponse(BaseModel):
"""Schema for order response."""
model_config = ConfigDict(from_attributes=True)
id: int
vendor_id: int
customer_id: int
order_number: str
# Channel/Source
channel: str
external_order_id: str | None = None
external_shipment_id: str | None = None
external_order_number: str | None = None
# Status
status: str
# Financial
subtotal: float | None
tax_amount: float | None
shipping_amount: float | None
discount_amount: float | None
total_amount: float
currency: str
# VAT information
vat_regime: str | None = None
vat_rate: float | None = None
vat_rate_label: str | None = None
vat_destination_country: str | None = None
# Customer snapshot
customer_first_name: str
customer_last_name: str
customer_email: str
customer_phone: str | None
customer_locale: str | None
# Shipping address snapshot
ship_first_name: str
ship_last_name: str
ship_company: str | None
ship_address_line_1: str
ship_address_line_2: str | None
ship_city: str
ship_postal_code: str
ship_country_iso: str
# Billing address snapshot
bill_first_name: str
bill_last_name: str
bill_company: str | None
bill_address_line_1: str
bill_address_line_2: str | None
bill_city: str
bill_postal_code: str
bill_country_iso: str
# Tracking
shipping_method: str | None
tracking_number: str | None
tracking_provider: str | None
tracking_url: str | None = None
shipment_number: str | None = None
shipping_carrier: str | None = None
# Notes
customer_notes: str | None
internal_notes: str | None
# Timestamps
order_date: datetime
confirmed_at: datetime | None
shipped_at: datetime | None
delivered_at: datetime | None
cancelled_at: datetime | None
created_at: datetime
updated_at: datetime
@property
def customer_full_name(self) -> str:
return f"{self.customer_first_name} {self.customer_last_name}".strip()
@property
def ship_full_name(self) -> str:
return f"{self.ship_first_name} {self.ship_last_name}".strip()
@property
def is_marketplace_order(self) -> bool:
return self.channel != "direct"
class OrderDetailResponse(OrderResponse):
"""Schema for detailed order response with items."""
items: list[OrderItemResponse] = []
# Vendor info (enriched by API)
vendor_name: str | None = None
vendor_code: str | None = None
class OrderListResponse(BaseModel):
"""Schema for paginated order list."""
orders: list[OrderResponse]
total: int
skip: int
limit: int
# ============================================================================
# Order List Item (Simplified for list views)
# ============================================================================
class OrderListItem(BaseModel):
"""Simplified order item for list views."""
model_config = ConfigDict(from_attributes=True)
id: int
vendor_id: int
order_number: str
channel: str
status: str
# External references
external_order_number: str | None = None
# Customer
customer_full_name: str
customer_email: str
# Financial
total_amount: float
currency: str
# Shipping
ship_country_iso: str
# Tracking
tracking_number: str | None
tracking_provider: str | None
tracking_url: str | None = None
shipment_number: str | None = None
shipping_carrier: str | None = None
# Item count
item_count: int = 0
# Timestamps
order_date: datetime
confirmed_at: datetime | None
shipped_at: datetime | None
# ============================================================================
# Admin Order Schemas
# ============================================================================
class AdminOrderItem(BaseModel):
"""Order item with vendor info for admin list view."""
model_config = ConfigDict(from_attributes=True)
id: int
vendor_id: int
vendor_name: str | None = None
vendor_code: str | None = None
customer_id: int
order_number: str
channel: str
status: str
# External references
external_order_number: str | None = None
external_shipment_id: str | None = None
# Customer snapshot
customer_full_name: str
customer_email: str
# Financial
subtotal: float | None
tax_amount: float | None
shipping_amount: float | None
discount_amount: float | None
total_amount: float
currency: str
# VAT information
vat_regime: str | None = None
vat_rate: float | None = None
vat_rate_label: str | None = None
vat_destination_country: str | None = None
# Shipping
ship_country_iso: str
tracking_number: str | None
tracking_provider: str | None
tracking_url: str | None = None
shipment_number: str | None = None
shipping_carrier: str | None = None
# Item count
item_count: int = 0
# Timestamps
order_date: datetime
confirmed_at: datetime | None
shipped_at: datetime | None
delivered_at: datetime | None
cancelled_at: datetime | None
created_at: datetime
updated_at: datetime
class AdminOrderListResponse(BaseModel):
"""Cross-vendor order list for admin."""
orders: list[AdminOrderItem]
total: int
skip: int
limit: int
class AdminOrderStats(BaseModel):
"""Order statistics for admin dashboard."""
total_orders: int = 0
pending_orders: int = 0
processing_orders: int = 0
shipped_orders: int = 0
delivered_orders: int = 0
cancelled_orders: int = 0
refunded_orders: int = 0
total_revenue: float = 0.0
# By channel
direct_orders: int = 0
letzshop_orders: int = 0
# Vendors
vendors_with_orders: int = 0
class AdminOrderStatusUpdate(BaseModel):
"""Admin version of status update with reason."""
status: str = Field(
..., pattern="^(pending|processing|shipped|delivered|cancelled|refunded)$"
)
tracking_number: str | None = None
tracking_provider: str | None = None
reason: str | None = Field(None, description="Reason for status change")
class AdminVendorWithOrders(BaseModel):
"""Vendor with order count."""
id: int
name: str
vendor_code: str
order_count: int = 0
class AdminVendorsWithOrdersResponse(BaseModel):
"""Response for vendors with orders list."""
vendors: list[AdminVendorWithOrders]
# ============================================================================
# Letzshop-specific Schemas
# ============================================================================
class LetzshopOrderImport(BaseModel):
"""Schema for importing a Letzshop order from shipment data."""
shipment_id: str
order_id: str
order_number: str
order_date: datetime
# Customer
customer_email: str
customer_locale: str | None = None
# Shipping address
ship_first_name: str
ship_last_name: str
ship_company: str | None = None
ship_address_line_1: str
ship_address_line_2: str | None = None
ship_city: str
ship_postal_code: str
ship_country_iso: str
# Billing address
bill_first_name: str
bill_last_name: str
bill_company: str | None = None
bill_address_line_1: str
bill_address_line_2: str | None = None
bill_city: str
bill_postal_code: str
bill_country_iso: str
# Totals
total_amount: float
currency: str = "EUR"
# State
letzshop_state: str # unconfirmed, confirmed, declined
# Items
inventory_units: list[dict]
# Raw data
raw_data: dict | None = None
class LetzshopShippingInfo(BaseModel):
"""Shipping info retrieved from Letzshop."""
tracking_number: str
tracking_provider: str
shipment_id: str
class LetzshopOrderConfirmItem(BaseModel):
"""Schema for confirming/declining a single item."""
item_id: int
external_item_id: str
action: str = Field(..., pattern="^(confirm|decline)$")
class LetzshopOrderConfirmRequest(BaseModel):
"""Schema for confirming/declining order items."""
items: list[LetzshopOrderConfirmItem]
# ============================================================================
# Mark as Shipped Schemas
# ============================================================================
class MarkAsShippedRequest(BaseModel):
"""Schema for marking an order as shipped with tracking info."""
tracking_number: str | None = Field(None, max_length=100)
tracking_url: str | None = Field(None, max_length=500)
shipping_carrier: str | None = Field(None, max_length=50)
class ShippingLabelInfo(BaseModel):
"""Shipping label information for an order."""
shipment_number: str | None = None
shipping_carrier: str | None = None
label_url: str | None = None
tracking_number: str | None = None
tracking_url: str | None = None
Reference in New Issue View Git Blame Copy Permalink