Spaces:
Configuration error
Configuration error
| """ | |
| 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 | |