mi
57 lines
1.8 KiB
Python
57 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
|
|
from app.domain import select_email
|
|
|
|
|
|
class UnknownCitizenship(ValueError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CitizenshipEntry:
|
|
enum_id: str
|
|
display_value: str
|
|
loaded_at: datetime
|
|
|
|
|
|
class CitizenshipDictionary:
|
|
def __init__(self, ttl_seconds: int = 3600) -> None:
|
|
self.ttl = timedelta(seconds=ttl_seconds)
|
|
self._entries: dict[str, CitizenshipEntry] = {}
|
|
self.loaded_at: datetime | None = None
|
|
|
|
def load(self, values: list[dict[str, Any]], now: datetime | None = None) -> None:
|
|
loaded_at = now or datetime.now(UTC)
|
|
self._entries = {
|
|
str(item["ID"]): CitizenshipEntry(
|
|
str(item["ID"]), str(item["VALUE"]), loaded_at
|
|
)
|
|
for item in values
|
|
if item.get("ID") is not None and isinstance(item.get("VALUE"), str)
|
|
}
|
|
self.loaded_at = loaded_at
|
|
|
|
def resolve(self, enum_id: str | int | None, now: datetime | None = None) -> str | None:
|
|
if enum_id in (None, ""):
|
|
return None
|
|
entry = self._entries.get(str(enum_id))
|
|
if entry is None:
|
|
raise UnknownCitizenship(str(enum_id))
|
|
if (now or datetime.now(UTC)) - entry.loaded_at > self.ttl:
|
|
raise UnknownCitizenship(str(enum_id))
|
|
return entry.display_value
|
|
|
|
|
|
def crm_master_projection(
|
|
contact: dict[str, Any], citizenship: CitizenshipDictionary, citizenship_field: str
|
|
) -> dict[str, str | None]:
|
|
return {
|
|
"full_name": contact.get("NAME") or None,
|
|
"email": select_email(contact.get("EMAIL") or []),
|
|
"citizenship": citizenship.resolve(contact.get(citizenship_field)),
|
|
}
|