MukeshKapoor25's picture
meta schema changes
00cd64a
"""
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