Spaces:
Running
Running
| """ | |
| 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 | |