89 lines
2.9 KiB
Python
89 lines
2.9 KiB
Python
"""Repository base exceptions + shared helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Mapping
|
|
from uuid import UUID
|
|
|
|
from motor.motor_asyncio import AsyncIOMotorCollection
|
|
from pydantic import BaseModel
|
|
|
|
from asf_domain.base import generate_id
|
|
|
|
|
|
class ASFRepositoryError(Exception):
|
|
"""Base for all repository-layer errors."""
|
|
|
|
|
|
class NotFoundError(ASFRepositoryError):
|
|
def __init__(self, kind: str, entity_id: UUID | str) -> None:
|
|
super().__init__(f"{kind} not found: {entity_id}")
|
|
self.kind = kind
|
|
self.entity_id = entity_id
|
|
|
|
|
|
class DuplicateError(ASFRepositoryError):
|
|
def __init__(self, kind: str, key: str) -> None:
|
|
super().__init__(f"{kind} already exists: {key}")
|
|
self.kind = kind
|
|
self.key = key
|
|
|
|
|
|
class TenantMismatchError(ASFRepositoryError):
|
|
"""Raised when an operation targets an entity owned by another tenant."""
|
|
|
|
def __init__(self, kind: str, entity_id: UUID | str) -> None:
|
|
super().__init__(f"Tenant isolation violation: {kind} {entity_id}")
|
|
|
|
|
|
def _now_utc() -> datetime:
|
|
return datetime.now(tz=timezone.utc)
|
|
|
|
|
|
def _to_doc(m: BaseModel) -> dict[str, Any]:
|
|
return m.model_dump(mode="python", by_alias=False)
|
|
|
|
|
|
def _from_doc(cls, doc: Mapping[str, Any]):
|
|
"""Instantiate a Pydantic model from a Mongo document.
|
|
|
|
``_id`` → ``id`` rename handled explicitly. Motor already returns UUID
|
|
objects for UUID fields when uuidRepresentation=standard is set.
|
|
|
|
Unknown document fields are filtered out (matches ``extra = forbid``
|
|
used by most ASF domain models while still tolerating new audit-only
|
|
Mongo columns like ``updated_at`` / ``metadata``).
|
|
"""
|
|
if doc is None:
|
|
return None
|
|
d = dict(doc)
|
|
if "_id" in d and "id" not in d:
|
|
d["id"] = d.pop("_id")
|
|
allowed = set(cls.model_fields.keys())
|
|
known = {k: v for k, v in d.items() if k in allowed}
|
|
try:
|
|
return cls.model_validate(known)
|
|
except Exception:
|
|
for k in list(known.keys()):
|
|
f = cls.model_fields.get(k)
|
|
if f is None:
|
|
continue
|
|
annotation = getattr(f, "annotation", None)
|
|
# Wrap None for required fields with usable defaults — job usage, etc.
|
|
if known[k] is None and annotation is not None:
|
|
from typing import get_origin, get_args
|
|
origin = get_origin(annotation)
|
|
args = get_args(annotation)
|
|
# Default dict for JobUsage
|
|
name = ""
|
|
if hasattr(annotation, "__name__"):
|
|
name = annotation.__name__
|
|
for arg in args:
|
|
if hasattr(arg, "__name__"):
|
|
name = arg.__name__
|
|
break
|
|
if name == "JobUsage":
|
|
known[k] = {"extra": {}}
|
|
return cls.model_validate(known)
|