""" Core utility functions for common operations across modules. """ from typing import Dict, Any, Optional def format_meta_field(document: Dict[str, Any]) -> Dict[str, Any]: """ Format a document by grouping audit fields under 'meta'. This utility transforms database documents to API response format by: 1. Extracting audit fields (created_by, created_by_username, etc.) 2. Creating a 'meta' object with proper field mapping 3. Removing audit fields from the top level 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 Args: document: Raw document from database with audit fields Returns: Formatted document with 'meta' field containing audit information Example: >>> doc = { ... "id": "123", ... "name": "Test", ... "created_by": "usr_admin", ... "created_by_username": "admin", ... "created_at": "2024-01-01T00:00:00Z" ... } >>> formatted = format_meta_field(doc) >>> formatted { "id": "123", "name": "Test", "meta": { "created_by": "admin", "created_by_id": "usr_admin", "created_at": "2024-01-01T00:00:00Z" } } """ # Create meta object from audit fields meta = { "created_by": document.get("created_by_username") or "Unknown", "created_by_id": str(document.get("created_by", "")), "created_at": document.get("created_at"), } # Add optional updated fields if they exist if "updated_by" in document and document["updated_by"]: meta["updated_by_id"] = str(document["updated_by"]) meta["updated_by"] = document.get("updated_by_username") or "Unknown" if "updated_at" in document and document["updated_at"]: meta["updated_at"] = document["updated_at"] # Create a copy of document without the audit fields audit_fields = [ "created_by", "created_at", "created_by_username", "updated_by", "updated_by_username", "updated_at" ] formatted = {k: v for k, v in document.items() if k not in audit_fields} # Add the meta field formatted["meta"] = meta return formatted def extract_audit_fields( user_id: str, username: Optional[str] = None, is_update: bool = False ) -> Dict[str, Any]: """ Extract audit fields for database operations. Creates a dictionary of audit fields to be stored in the database. Use this when creating or updating documents. Args: user_id: UUID of the user performing the operation username: Username of the user (optional, defaults to "Unknown") is_update: If True, returns update fields; if False, returns creation fields Returns: Dictionary with audit fields ready for database storage Example: >>> # For creation >>> audit = extract_audit_fields("usr_123", "admin", is_update=False) >>> audit { "created_by": "usr_123", "created_by_username": "admin", "created_at": datetime(...) } >>> # For update >>> audit = extract_audit_fields("usr_456", "manager", is_update=True) >>> audit { "updated_by": "usr_456", "updated_by_username": "manager", "updated_at": datetime(...) } """ from datetime import datetime if is_update: return { "updated_by": user_id, "updated_by_username": username or "Unknown", "updated_at": datetime.utcnow() } else: return { "created_by": user_id, "created_by_username": username or "Unknown", "created_at": datetime.utcnow() } def normalize_uuid_fields(document: Dict[str, Any], fields: list[str]) -> Dict[str, Any]: """ Normalize UUID fields to strings in a document. Converts UUID objects to strings for fields specified in the fields list. Useful when preparing documents for JSON serialization. Args: document: Document with potential UUID fields fields: List of field names that should be converted to strings Returns: Document with UUID fields converted to strings Example: >>> from uuid import UUID >>> doc = {"user_id": UUID("123e4567-e89b-12d3-a456-426614174000"), "name": "Test"} >>> normalized = normalize_uuid_fields(doc, ["user_id"]) >>> normalized["user_id"] "123e4567-e89b-12d3-a456-426614174000" """ for field in fields: if field in document and document[field] is not None: document[field] = str(document[field]) return document