app/core/README.md

Core Schemas and Utilities

This directory contains reusable schemas and utility functions that can be used across all modules in the SCM microservice.

Files

schemas.py

Contains common Pydantic schemas for consistent API responses.

utils.py

Contains utility functions for common operations like formatting meta fields.

Available Schemas

1. MetaSchema

Groups audit information in a consistent structure across all entities.

Usage in Response Schemas:

from app.core.schemas import MetaSchema

class MerchantResponse(BaseModel):
    merchant_id: str
    name: str
    # ... other fields
    meta: Dict[str, Any]  # Will contain MetaSchema structure

Fields:

• created_by: Username who created the record (human-readable)
• created_by_id: UUID who created the record
• created_at: Creation timestamp
• updated_by: Username who last updated (optional)
• updated_by_id: UUID who last updated (optional)
• updated_at: Last update timestamp (optional)

Example Response:

{
  "merchant_id": "mrc_123",
  "name": "ABC Store",
  "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"
  }
}

Meta Schema Usage by Entity

The following entity responses must include the meta object as defined by MetaSchema.

Merchant Meta Schema

Applies to: MerchantResponse, MerchantListResponse items

Required meta fields:

• created_by
• created_by_id
• created_at

Optional meta fields:

• updated_by
• updated_by_id
• updated_at

Example:

{
    "merchant_id": "mrc_123",
    "name": "ABC Store",
    "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"
    }
}

Employee Meta Schema

Applies to: EmployeeResponse, EmployeeListResponse items

Required meta fields:

• created_by
• created_by_id
• created_at

Optional meta fields:

• updated_by
• updated_by_id
• updated_at

Example:

{
    "employee_id": "emp_456",
    "full_name": "Jane Doe",
    "meta": {
        "created_by": "admin",
        "created_by_id": "usr_01HZQX5K3N2P8R6T4V9W",
        "created_at": "2023-03-11T12:15:00Z",
        "updated_by": "hr_manager",
        "updated_by_id": "usr_01HZQX5K3N2P8R6T4V9X",
        "updated_at": "2024-05-02T09:30:00Z"
    }
}

Staff Meta Schema

Applies to: StaffResponse, StaffListResponse items

Required meta fields:

• created_by
• created_by_id
• created_at

Optional meta fields:

• updated_by
• updated_by_id
• updated_at

Example:

{
    "staff_id": "stf_789",
    "name": "Rahul Singh",
    "meta": {
        "created_by": "admin",
        "created_by_id": "usr_01HZQX5K3N2P8R6T4V9W",
        "created_at": "2023-06-01T07:45:00Z",
        "updated_by": "ops_manager",
        "updated_by_id": "usr_01HZQX5K3N2P8R6T4V9X",
        "updated_at": "2024-02-19T18:10:00Z"
    }
}

Customer Meta Schema

Applies to: CustomerResponse, CustomerListResponse items

Required meta fields:

• created_by
• created_by_id
• created_at

Optional meta fields:

• updated_by
• updated_by_id
• updated_at

Example:

{
    "customer_id": "cus_012",
    "full_name": "Ayesha Khan",
    "meta": {
        "created_by": "admin",
        "created_by_id": "usr_01HZQX5K3N2P8R6T4V9W",
        "created_at": "2023-08-22T14:05:00Z",
        "updated_by": "support_agent",
        "updated_by_id": "usr_01HZQX5K3N2P8R6T4V9X",
        "updated_at": "2024-01-16T16:20:00Z"
    }
}

2. PaginationMeta

Provides pagination information for list endpoints.

Usage:

from app.core.schemas import PaginationMeta

class MerchantListResponse(BaseModel):
    data: List[MerchantResponse]
    pagination: PaginationMeta

3. ErrorDetail

Standard error detail structure for validation errors.

Usage:

from app.core.schemas import ErrorDetail

class ValidationErrorResponse(BaseModel):
    errors: List[ErrorDetail]

4. SuccessResponse

Standard success response for operations without data.

Usage:

from app.core.schemas import SuccessResponse

@router.delete("/{id}")
async def delete_item(id: str) -> SuccessResponse:
    # ... delete logic
    return SuccessResponse(message="Item deleted successfully")

Available Utilities

1. format_meta_field()

Transforms database documents to API response format by grouping audit fields under 'meta'.

Usage in Service Layer:

from app.core.utils import format_meta_field

class MerchantService:
    @staticmethod
    async def get_merchant(merchant_id: str) -> MerchantResponse:
        # Fetch from database
        merchant = await db.merchants.find_one({"merchant_id": merchant_id})
        
        # Format with meta field
        formatted = format_meta_field(merchant)
        
        # Return response
        return MerchantResponse(**formatted)

What it does:

• Extracts audit fields from document
• Creates 'meta' object with proper field mapping
• Removes audit fields from top level
• Returns formatted document

Field Mapping:

DB 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

2. extract_audit_fields()

Creates audit field dictionaries for database operations.

Usage for Creation:

from app.core.utils import extract_audit_fields

# When creating a new record
audit_fields = extract_audit_fields(
    user_id=current_user.user_id,
    username=current_user.username,
    is_update=False
)

merchant_data = {
    "merchant_id": "mrc_123",
    "name": "ABC Store",
    **audit_fields  # Adds created_by, created_by_username, created_at
}

Usage for Updates:

# When updating a record
audit_fields = extract_audit_fields(
    user_id=current_user.user_id,
    username=current_user.username,
    is_update=True
)

update_data = {
    "name": "New Name",
    **audit_fields  # Adds updated_by, updated_by_username, updated_at
}

3. normalize_uuid_fields()

Converts UUID objects to strings for JSON serialization.

Usage:

from app.core.utils import normalize_uuid_fields

# After fetching from database
document = await db.collection.find_one({"id": some_uuid})

# Normalize UUID fields
normalized = normalize_uuid_fields(
    document, 
    fields=["user_id", "manager_id", "created_by"]
)

Implementation Guide for New Modules

Step 1: Update Model

Add audit fields to your MongoDB model:

from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional

class MerchantModel(BaseModel):
    merchant_id: str
    name: str
    # ... other 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: Optional[datetime] = Field(None)

Step 2: Update Response Schema

Import MetaSchema and use it in your response:

from typing import Dict, Any
from pydantic import BaseModel
from app.core.schemas import MetaSchema

class MerchantResponse(BaseModel):
    merchant_id: str
    name: str
    # ... other fields
    meta: Dict[str, Any]  # Contains MetaSchema structure

Step 3: Update Service Layer

Use the utility functions in your service:

from app.core.utils import format_meta_field, extract_audit_fields

class MerchantService:
    @staticmethod
    async def create_merchant(payload: MerchantCreate, current_user: TokenUser):
        # Add audit fields for creation
        audit_fields = extract_audit_fields(
            user_id=current_user.user_id,
            username=current_user.username,
            is_update=False
        )
        
        merchant_data = {
            **payload.dict(),
            **audit_fields
        }
        
        # Insert into database
        await db.merchants.insert_one(merchant_data)
        
        # Format response with meta
        formatted = format_meta_field(merchant_data)
        return MerchantResponse(**formatted)
    
    @staticmethod
    async def update_merchant(
        merchant_id: str, 
        payload: MerchantUpdate,
        current_user: TokenUser
    ):
        # Add audit fields for update
        audit_fields = extract_audit_fields(
            user_id=current_user.user_id,
            username=current_user.username,
            is_update=True
        )
        
        update_data = {
            **payload.dict(exclude_unset=True),
            **audit_fields
        }
        
        # Update in database
        await db.merchants.update_one(
            {"merchant_id": merchant_id},
            {"$set": update_data}
        )
        
        # Fetch and format response
        merchant = await db.merchants.find_one({"merchant_id": merchant_id})
        formatted = format_meta_field(merchant)
        return MerchantResponse(**formatted)
    
    @staticmethod
    async def get_merchant(merchant_id: str):
        merchant = await db.merchants.find_one({"merchant_id": merchant_id})
        if not merchant:
            raise HTTPException(status_code=404, detail="Merchant not found")
        
        # Format with meta
        formatted = format_meta_field(merchant)
        return MerchantResponse(**formatted)
    
    @staticmethod
    async def list_merchants(skip: int = 0, limit: int = 100):
        cursor = db.merchants.find().skip(skip).limit(limit)
        merchants = await cursor.to_list(length=limit)
        
        # Format each merchant with meta
        formatted_merchants = [
            MerchantResponse(**format_meta_field(m)) 
            for m in merchants
        ]
        
        return formatted_merchants

Step 4: Update Controller

Ensure controllers pass both user_id and username:

@router.post("/")
async def create_merchant(
    payload: MerchantCreate,
    current_user: TokenUser = Depends(get_current_user)
):
    return await MerchantService.create_merchant(payload, current_user)

@router.patch("/{merchant_id}")
async def update_merchant(
    merchant_id: str,
    payload: MerchantUpdate,
    current_user: TokenUser = Depends(get_current_user)
):
    return await MerchantService.update_merchant(
        merchant_id, 
        payload, 
        current_user
    )

Benefits

1. Consistency: All entities use the same meta structure
2. Reusability: Write once, use everywhere
3. Maintainability: Changes to meta structure happen in one place
4. Type Safety: Pydantic validation ensures correct structure
5. Documentation: Self-documenting with clear field descriptions

Migration Checklist

When migrating an existing module to use MetaSchema:

• Add audit fields to model (if not present)
• Import MetaSchema in response schema
• Update response schema to use meta: Dict[str, Any]
• Import utility functions in service
• Update create method to use extract_audit_fields(is_update=False)
• Update update method to use extract_audit_fields(is_update=True)
• Update all methods returning responses to use format_meta_field()
• Update controller to pass current_user to service methods
• Test all CRUD operations
• Update API documentation

Examples

See app/employees/ for a complete implementation example.