cuatrolabs-pos-ms / app /core /schemas.py
MukeshKapoor25's picture
feat: Enhance audit trail with created_by and updated_by fields across services
8a996cb
"""
Core/Common Pydantic schemas used across multiple modules.
These schemas provide reusable components for consistent API responses.
"""
from typing import Optional, Any, Dict, List
from datetime import datetime
from pydantic import BaseModel, Field
class MetaSchema(BaseModel):
"""
Audit and metadata information for tracking record lifecycle.
This schema groups all audit-related fields in a consistent structure
that can be used across all entities (customers, staff, appointments, etc.).
Fields:
created_by: Username of the user who created the record (human-readable)
created_by_id: UUID of the user who created the record
created_at: Timestamp when the record was created
updated_by: Username of the user who last updated the record (optional)
updated_by_id: UUID of the user who last updated the record (optional)
updated_at: Timestamp when the record was last updated (optional)
"""
created_by: str = Field(..., description="Username who created this record")
created_by_id: str = Field(..., description="User ID (UUID) who created this record")
created_at: datetime = Field(..., description="Timestamp when record was created")
updated_by: Optional[str] = Field(None, description="Username who last updated this record")
updated_by_id: Optional[str] = Field(None, description="User ID (UUID) who last updated this record")
updated_at: Optional[datetime] = Field(None, description="Timestamp when record was last updated")
class Config:
json_schema_extra = {
"example": {
"created_by": "admin",
"created_by_id": "usr_01HZQX5K3N2P8R6T4V9W",
"created_at": "2023-01-10T08:00:00Z",
"updated_by": "manager",
"updated_by_id": "usr_01HZQX5K3N2P8R6T4V9X",
"updated_at": "2024-11-24T10:00:00Z"
}
}
class PaginationMeta(BaseModel):
"""
Pagination metadata for list endpoints.
Provides information about the current page, total records, and pagination state.
"""
total: int = Field(..., description="Total number of records matching the query")
skip: int = Field(..., description="Number of records skipped (offset)")
limit: int = Field(..., description="Maximum number of records returned")
has_more: bool = Field(..., description="Whether there are more records available")
class Config:
json_schema_extra = {
"example": {
"total": 150,
"skip": 0,
"limit": 100,
"has_more": True
}
}
class ErrorDetail(BaseModel):
"""
Detailed error information for validation and business logic errors.
"""
field: str = Field(..., description="Field name that caused the error")
message: str = Field(..., description="Human-readable error message")
code: Optional[str] = Field(None, description="Machine-readable error code")
class Config:
json_schema_extra = {
"example": {
"field": "email",
"message": "Email address is already registered",
"code": "DUPLICATE_EMAIL"
}
}
class SuccessResponse(BaseModel):
"""
Standard success response for operations that don't return data.
"""
success: bool = Field(True, description="Operation success status")
message: str = Field(..., description="Success message")
class Config:
json_schema_extra = {
"example": {
"success": True,
"message": "Operation completed successfully"
}
}
class StatusResponse(BaseModel):
"""
Standard status response for operations with optional data payload.
Used for API endpoints that return status with additional data.
"""
success: bool = Field(..., description="Operation success status")
message: str = Field(..., description="Status or success message")
data: Optional[Dict[str, Any]] = Field(None, description="Optional data payload")
class Config:
json_schema_extra = {
"example": {
"success": True,
"message": "Operation completed successfully",
"data": {
"count": 10,
"items": []
}
}
}
class BaseLazyFetchSchema(BaseModel):
"""
Base schema for lazy-fetch list requests with pagination and projection support.
Provides common fields for listing endpoints with filtering, pagination, and field projection.
"""
skip: int = Field(0, ge=0, description="Number of records to skip (pagination offset)")
limit: int = Field(100, ge=1, le=500, description="Maximum number of records to return")
projection_list: Optional[List[str]] = Field(
None,
description="List of fields to include in response (omit for all fields)"
)
class Config:
json_schema_extra = {
"example": {
"skip": 0,
"limit": 100,
"projection_list": ["id", "name", "status"]
}
}