Spaces:
Configuration error
Configuration error
File size: 2,059 Bytes
3818b61 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 | """
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
|