POS_META_SCHEMA_IMPLEMENTATION.md

POS Microservice: Meta Schema Implementation

Overview

This document describes the implementation of the Meta Schema Pattern in the POS microservice. This pattern groups audit fields (created_by, created_at, updated_by, updated_at) under a single 'meta' object in API responses, providing consistent audit trail information across all entities.

Implementation Date

December 2024

Modules Updated

1. Customers Module (app/customers/)
2. Staff Module (app/staff/)

Core Infrastructure

1. Core Schemas (app/core/schemas.py)

Added the following reusable schemas:

• MetaSchema: Defines the structure for audit information
• PaginationMeta: Pagination information for list endpoints
• ErrorDetail: Standard error detail structure
• SuccessResponse: Standard success response
• StatusResponse: Standard status response with optional data payload
• BaseLazyFetchSchema: Base schema for list requests with pagination

2. Core Utilities (app/core/utils.py)

Three essential utility functions:

format_meta_field(document: dict) -> dict

Transforms database documents to API response format by:

• Extracting audit fields from document
• Creating 'meta' object with proper field mapping
• Removing audit fields from top level
• Returning formatted document

Field Mapping:

Database Field          → Response Field
----------------          ----------------
created_by (UUID)       → meta.created_by_id
created_by_username     → meta.created_by
created_at              → meta.created_at
updated_by (UUID)       → meta.updated_by_id
updated_by_username     → meta.updated_by
updated_at              → meta.updated_at

extract_audit_fields(user_id: str, username: str, is_update: bool) -> dict

Creates audit field dictionaries for database operations:

• For creation (is_update=False): Returns created_by, created_by_username, created_at
• For updates (is_update=True): Returns updated_by, updated_by_username, updated_at

normalize_uuid_fields(document: dict, fields: List[str]) -> dict

Converts UUID objects to strings for JSON serialization.

3. Core Documentation (app/core/README.md)

Comprehensive documentation with:

• Usage examples for all entities
• Implementation guide for new modules
• Benefits and best practices
• Migration checklist

Customers Module Implementation

Changes Made

1. Model (app/customers/models/model.py)

Added Audit Fields:

class CustomerModel(BaseModel):
    # ... existing fields ...
    
    # Audit fields
    created_by: str = Field(..., description="User ID who created")
    created_by_username: Optional[str] = Field(None, description="Username who created")
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_by: Optional[str] = Field(None, description="User ID who updated")
    updated_by_username: Optional[str] = Field(None, description="Username who updated")
    updated_at: datetime = Field(default_factory=datetime.utcnow)

2. Schemas (app/customers/schemas/schema.py)

Updated CustomerCreate:

class CustomerCreate(CustomerBase):
    merchant_id: Optional[str] = Field(None, description="Merchant identifier (UUID) - will be overridden by token")
    created_by_username: Optional[str] = Field(None, description="Username who created - will be set from token")

Updated CustomerUpdate:

class CustomerUpdate(BaseModel):
    # ... existing fields ...
    updated_by_username: Optional[str] = Field(None, description="Username who updated - will be set from token")

Updated CustomerResponse:

class CustomerResponse(CustomerBase):
    customer_id: str = Field(..., description="Customer identifier (UUID)")
    meta: Dict[str, Any] = Field(..., description="Audit information (created_by, created_at, updated_by, updated_at)")
    
    class Config:
        from_attributes = True

3. Service (app/customers/services/service.py)

Added Imports:

from app.core.utils import format_meta_field, extract_audit_fields

Updated create_customer:

• Added user_id and username parameters
• Uses extract_audit_fields(is_update=False) to add audit fields
• Uses format_meta_field() before returning CustomerResponse

Updated get_customer:

• Uses format_meta_field() before returning CustomerResponse

Updated list_customers:

• Uses format_meta_field() for each document when not using projection
• Returns raw dicts when projection is used (for performance)

Updated update_customer:

• Added user_id and username parameters
• Uses extract_audit_fields(is_update=True) to add update audit fields
• Uses format_meta_field() before returning CustomerResponse

Updated delete_customer:

• Added user_id and username parameters
• Uses extract_audit_fields(is_update=True) for soft delete audit trail
• Uses format_meta_field() before returning CustomerResponse

4. Controller (app/customers/controllers/router.py)

Updated create_customer endpoint:

async def create_customer(
    payload: CustomerCreate,
    current_user: TokenUser = Depends(require_pos_permission("customers", "create"))
):
    payload.created_by_username = current_user.username
    customer = await CustomerService.create_customer(
        payload, 
        current_user.merchant_id,
        current_user.user_id,
        current_user.username
    )

Updated update_customer endpoint:

async def update_customer(
    customer_id: str, 
    payload: CustomerUpdate,
    current_user: TokenUser = Depends(require_pos_permission("customers", "update"))
):
    payload.updated_by_username = current_user.username
    customer = await CustomerService.update_customer(
        customer_id=customer_id,
        payload=payload,
        merchant_id=current_user.merchant_id,
        merchant_type=current_user.merchant_type,
        user_id=current_user.user_id,
        username=current_user.username
    )

Updated delete_customer endpoint:

async def delete_customer(
    customer_id: str,
    current_user: TokenUser = Depends(require_pos_permission("customers", "delete"))
):
    await CustomerService.delete_customer(
        customer_id=customer_id,
        merchant_id=current_user.merchant_id,
        merchant_type=current_user.merchant_type,
        user_id=current_user.user_id,
        username=current_user.username
    )

API Response Example (Customers)

Before (Old Format):

{
  "customer_id": "cus_123",
  "name": "John Doe",
  "phone": "+91-9988776655",
  "email": "john@example.com",
  "status": "active",
  "created_at": "2023-01-10T08:00:00Z",
  "updated_at": "2024-11-24T10:00:00Z"
}

After (Meta Format):

{
  "customer_id": "cus_123",
  "name": "John Doe",
  "phone": "+91-9988776655",
  "email": "john@example.com",
  "status": "active",
  "meta": {
    "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"
  }
}

Staff Module Implementation

Changes Made

1. Model (app/staff/models/staff_model.py)

Added Audit Fields:

class StaffModel(BaseModel):
    # ... existing fields ...
    
    # Audit fields
    created_by: str = Field(..., description="User ID who created")
    created_by_username: Optional[str] = Field(None, description="Username who created")
    created_at: datetime = Field(default_factory=datetime.utcnow, description="Creation timestamp")
    updated_by: Optional[str] = Field(None, description="User ID who updated")
    updated_by_username: Optional[str] = Field(None, description="Username who updated")
    updated_at: datetime = Field(default_factory=datetime.utcnow, description="Last update timestamp")

2. Schemas (app/staff/schemas/staff_schema.py)

Updated StaffCreateSchema:

class StaffCreateSchema(BaseModel):
    # ... existing fields ...
    created_by_username: Optional[str] = Field(None, description="Username who created - will be set from token")

Updated StaffUpdateSchema:

class StaffUpdateSchema(BaseModel):
    # ... existing fields ...
    updated_by_username: Optional[str] = Field(None, description="Username who updated - will be set from token")

Updated StaffResponseSchema:

class StaffResponseSchema(BaseModel):
    staff_id: UUID
    merchant_id: UUID
    name: str
    role: str
    # ... other fields ...
    meta: dict = Field(..., description="Audit information (created_by, created_at, updated_by, updated_at)")
    
    class Config:
        from_attributes = True

3. Service (app/staff/services/staff_service.py)

Added Imports:

from app.core.utils import format_meta_field, extract_audit_fields

Updated create_staff:

• Added user_id and username parameters
• Uses extract_audit_fields(is_update=False) to add audit fields
• Uses format_meta_field() before returning StaffResponseSchema

Updated get_staff_by_id:

• Uses format_meta_field() before returning StaffResponseSchema

Updated update_staff:

• Added user_id and username parameters
• Uses extract_audit_fields(is_update=True) to add update audit fields
• Uses format_meta_field() before returning StaffResponseSchema (via get_staff_by_id)

Note: list_staff returns raw dicts and doesn't need formatting (projection-based approach).

4. Controller (app/staff/controllers/router.py)

Updated create_staff endpoint:

async def create_staff(
    payload: StaffCreateSchema,
    current_user: TokenUser = Depends(require_pos_permission("staff", "create"))
):
    payload.created_by_username = current_user.username
    result = await StaffService.create_staff(
        payload,
        user_id=current_user.user_id,
        username=current_user.username
    )

Updated update_staff endpoint:

async def update_staff(
    staff_id: UUID,
    payload: StaffUpdateSchema,
    current_user: TokenUser = Depends(require_pos_permission("staff", "edit"))
):
    payload.updated_by_username = current_user.username
    result = await StaffService.update_staff(
        staff_id, 
        current_user.merchant_id, 
        payload,
        user_id=current_user.user_id,
        username=current_user.username
    )

Updated update_staff_status endpoint:

async def update_staff_status(
    staff_id: UUID,
    payload: UpdateStatusRequest,
    current_user: TokenUser = Depends(require_pos_permission("staff", "edit"))
):
    update_payload = StaffUpdateSchema(
        status=payload.status,
        updated_by_username=current_user.username
    )
    result = await StaffService.update_staff(
        staff_id, 
        current_user.merchant_id, 
        update_payload,
        user_id=current_user.user_id,
        username=current_user.username
    )

API Response Example (Staff)

Before (Old Format):

{
  "staff_id": "staff_123",
  "merchant_id": "merchant_789",
  "name": "Aarav Sharma",
  "role": "Stylist",
  "phone": "+91-9876543210",
  "email": "aarav@retail.com",
  "status": "active",
  "created_at": "2024-01-15T10:30:00Z",
  "updated_at": "2024-01-15T10:30:00Z"
}

After (Meta Format):

{
  "staff_id": "staff_123",
  "merchant_id": "merchant_789",
  "name": "Aarav Sharma",
  "role": "Stylist",
  "phone": "+91-9876543210",
  "email": "aarav@retail.com",
  "status": "active",
  "meta": {
    "created_by": "admin",
    "created_by_id": "usr_01HZQX5K3N2P8R6T4V9W",
    "created_at": "2024-01-15T10:30:00Z",
    "updated_by": "manager",
    "updated_by_id": "usr_01HZQX5K3N2P8R6T4V9X",
    "updated_at": "2024-12-01T14:20:00Z"
  }
}

Benefits

1. Consistency

All entities (customers, staff, appointments, sales, etc.) use the same meta structure for audit information.

2. Human-Readable Audit Trail

Response includes both username (human-readable) and user_id (for system operations):

• meta.created_by: "admin" (easy to understand)
• meta.created_by_id: "usr_01HZQX..." (for system lookups)

3. Clean API Responses

Audit fields are grouped under 'meta', reducing clutter in the main response body.

4. Reusability

Utilities are in app/core/utils.py and can be used across all modules without duplication.

5. Type Safety

Pydantic schemas ensure correct structure and validation.

6. Database Flexibility

Database documents maintain flat structure (good for querying), while API responses use nested structure (good for clarity).

Implementation Pattern Summary

For any new module or entity, follow this pattern:

Step 1: Update Model

# Add audit fields to MongoDB model
created_by: str
created_by_username: Optional[str]
created_at: datetime
updated_by: Optional[str]
updated_by_username: Optional[str]
updated_at: datetime

Step 2: Update Schemas

# Create schema: add created_by_username
class EntityCreate(BaseModel):
    created_by_username: Optional[str] = None

# Update schema: add updated_by_username
class EntityUpdate(BaseModel):
    updated_by_username: Optional[str] = None

# Response schema: replace timestamps with meta
class EntityResponse(BaseModel):
    meta: Dict[str, Any]  # Instead of created_at, updated_at

Step 3: Update Service

from app.core.utils import format_meta_field, extract_audit_fields

# In create method
async def create_entity(payload, user_id, username):
    data = payload.dict()
    audit_fields = extract_audit_fields(user_id, username, is_update=False)
    data.update(audit_fields)
    # ... insert into DB ...
    formatted = format_meta_field(data)
    return EntityResponse(**formatted)

# In update method
async def update_entity(entity_id, payload, user_id, username):
    update_data = payload.dict(exclude_unset=True)
    audit_fields = extract_audit_fields(user_id, username, is_update=True)
    update_data.update(audit_fields)
    # ... update in DB ...
    doc = await get_entity(entity_id)
    formatted = format_meta_field(doc)
    return EntityResponse(**formatted)

# In get/list methods
async def get_entity(entity_id):
    doc = await db.find_one({"entity_id": entity_id})
    formatted = format_meta_field(doc)
    return EntityResponse(**formatted)

Step 4: Update Controller

# Pass user info from token to service
@router.post("/")
async def create_entity(
    payload: EntityCreate,
    current_user: TokenUser = Depends(get_current_user)
):
    payload.created_by_username = current_user.username
    return await EntityService.create_entity(
        payload,
        user_id=current_user.user_id,
        username=current_user.username
    )

@router.put("/{entity_id}")
async def update_entity(
    entity_id: str,
    payload: EntityUpdate,
    current_user: TokenUser = Depends(get_current_user)
):
    payload.updated_by_username = current_user.username
    return await EntityService.update_entity(
        entity_id,
        payload,
        user_id=current_user.user_id,
        username=current_user.username
    )

Testing Checklist

For each updated module:

• Test POST (create) endpoint - verify meta.created_by and meta.created_at
• Test GET (retrieve) endpoint - verify meta structure
• Test PUT/PATCH (update) endpoint - verify meta.updated_by and meta.updated_at
• Test DELETE (soft delete) endpoint - verify meta.updated_by
• Test list endpoint without projection - verify meta in each item
• Test list endpoint with projection - verify raw fields returned
• Verify database documents maintain flat structure
• Verify API responses use nested meta structure

Migration Notes

Backward Compatibility

• Database: Existing documents without audit fields will still work (fields are optional in get operations)
• API: Responses now include 'meta' instead of top-level created_at/updated_at
• Clients: Frontend/mobile apps will need updates to read from meta object

Database Migration

No migration needed for existing documents. New documents will have audit fields, old documents will work without them (format_meta_field handles missing fields gracefully).

Rollback Plan

If needed, can revert by:

1. Removing meta field from response schemas
2. Adding back created_at/updated_at to response schemas
3. Removing format_meta_field calls from services
4. Removing extract_audit_fields calls from create/update operations

Future Enhancements

Potential Additions

1. Deletion Audit: Add deleted_by, deleted_by_username, deleted_at for hard deletes
2. Version Tracking: Add version number to meta for optimistic locking
3. Change History: Store full change history in separate collection
4. IP Address Tracking: Add request_ip to audit fields
5. Reason Tracking: Add change_reason field for important updates

Other Modules to Update

• Appointments Module: Apply same pattern
• Sales Module: Apply same pattern
• Wallet Module: Apply same pattern
• Inventory Module (if exists): Apply same pattern

Related Documentation

• Core README - Detailed usage guide for core schemas and utilities
• SCM Meta Implementation - Similar implementation in SCM microservice

Contact

For questions or issues with this implementation, contact the backend team.

Version History

• v1.0 (December 2024): Initial implementation for Customers and Staff modules