""" Core utility functions shared across modules. """ from datetime import datetime from typing import Dict, Any, Optional, List def format_meta_field(document: Dict[str, Any]) -> Dict[str, Any]: """Group audit fields under 'meta' for API responses.""" meta = { "created_by": document.get("created_by_username") or "Unknown", "created_by_id": str(document.get("created_by", "")), "created_at": document.get("created_at"), } if document.get("updated_by"): meta["updated_by_id"] = str(document["updated_by"]) meta["updated_by"] = document.get("updated_by_username") or "Unknown" if document.get("updated_at"): meta["updated_at"] = document["updated_at"] 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} formatted["meta"] = meta formatted["created_by"] = document.get("created_by_username") or str(document.get("created_by", "")) formatted["created_at"] = document.get("created_at") if "updated_at" in document: formatted["updated_at"] = document.get("updated_at") return formatted def extract_audit_fields( user_id: str, username: Optional[str] = None, is_update: bool = False ) -> Dict[str, Any]: """Return audit fields dict for DB insert/update.""" if is_update: return { "updated_by": user_id, "updated_by_username": username or "Unknown", "updated_at": datetime.utcnow(), } return { "created_by": user_id, "created_by_username": username or "Unknown", "created_at": datetime.utcnow(), } def build_projection(projection_list: Optional[List[str]]) -> Optional[Dict[str, int]]: """Build a MongoDB projection dict from a list of field names.""" if not projection_list: return None proj = {field: 1 for field in projection_list} proj["_id"] = 0 return proj