File size: 1,496 Bytes
69f2337 a08f988 69f2337 a08f988 69f2337 a08f988 69f2337 |
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 |
from typing import Any, Callable
from uuid import UUID
from fastapi import HTTPException, status
from pydantic import BaseModel, Field
from app.core.config import TABLE_PARTICIPANTS, TABLE_TOPICS
class Link(BaseModel):
table: str
value: UUID
def resolve(self, resolvers: dict[str, Callable[[UUID], Any]]) -> Any:
resolver = resolvers.get(self.table)
if resolver is None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Resolver for table '{self.table}' is not registered",
)
return resolver(self.value)
def as_path(self, leading_slash: bool = True) -> str:
path = f"{self.table}/{self.value}"
return f"/{path}" if leading_slash else path
@classmethod
def from_raw(cls, raw: str) -> "Link":
text = raw.strip()
if not text:
raise ValueError("Cannot parse empty link value")
normalized = text.lstrip("/")
table, value = normalized.split("/", 1)
link_cls = LINK_TYPE_REGISTRY.get(table, Link)
if link_cls is Link:
return link_cls(table=table, value=UUID(value))
return link_cls(value=UUID(value))
class TopicLink(Link):
table: str = Field(default=TABLE_TOPICS)
class ParticipantLink(Link):
table: str = Field(default=TABLE_PARTICIPANTS)
LINK_TYPE_REGISTRY: dict[str, type[Link]] = {
TABLE_TOPICS: TopicLink,
TABLE_PARTICIPANTS: ParticipantLink,
}
|