|
|
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, |
|
|
} |
|
|
|
|
|
|
