rose-ash/shared/contracts/dtos.py

"""Frozen dataclasses for cross-domain data transfer.

These are the *only* shapes that cross domain boundaries.  Consumers never
see ORM model instances from another domain — only these DTOs.
"""
from __future__ import annotations

import dataclasses
import typing
from dataclasses import dataclass, field
from datetime import datetime
from decimal import Decimal


# ---------------------------------------------------------------------------
# Serialization helpers for JSON transport over internal data endpoints
# ---------------------------------------------------------------------------

def _serialize_value(v):
    """Convert a single value to a JSON-safe type."""
    if isinstance(v, datetime):
        return v.isoformat()
    if isinstance(v, Decimal):
        return str(v)
    if isinstance(v, set):
        return list(v)
    if dataclasses.is_dataclass(v) and not isinstance(v, type):
        return dto_to_dict(v)
    if isinstance(v, list):
        return [_serialize_value(item) for item in v]
    return v


def dto_to_dict(obj) -> dict:
    """Convert a frozen DTO dataclass to a JSON-serialisable dict."""
    return {k: _serialize_value(v) for k, v in dataclasses.asdict(obj).items()}


def _unwrap_optional(hint):
    """Unwrap Optional[X] / X | None to return X."""
    args = getattr(hint, "__args__", ())
    if args:
        real = [a for a in args if a is not type(None)]
        if real:
            return real[0]
    return hint


def dto_from_dict(cls, data: dict):
    """Construct a DTO from a dict, coercing dates and Decimals.

    Uses ``typing.get_type_hints()`` to resolve forward-ref annotations
    (from ``from __future__ import annotations``).
    """
    if not data:
        return None
    try:
        hints = typing.get_type_hints(cls)
    except Exception:
        hints = {}
    kwargs = {}
    for f in dataclasses.fields(cls):
        if f.name not in data:
            continue
        val = data[f.name]
        if val is not None and f.name in hints:
            hint = _unwrap_optional(hints[f.name])
            if hint is datetime and isinstance(val, str):
                val = datetime.fromisoformat(val)
            elif hint is Decimal:
                val = Decimal(str(val))
        kwargs[f.name] = val
    return cls(**kwargs)


# ---------------------------------------------------------------------------
# Blog domain
# ---------------------------------------------------------------------------

@dataclass(frozen=True, slots=True)
class PostDTO:
    id: int
    slug: str
    title: str
    status: str
    visibility: str
    is_page: bool = False
    feature_image: str | None = None
    html: str | None = None
    sx_content: str | None = None
    excerpt: str | None = None
    custom_excerpt: str | None = None
    published_at: datetime | None = None


# ---------------------------------------------------------------------------
# Calendar / Events domain
# ---------------------------------------------------------------------------

@dataclass(frozen=True, slots=True)
class CalendarDTO:
    id: int
    container_type: str
    container_id: int
    name: str
    slug: str
    description: str | None = None


@dataclass(frozen=True, slots=True)
class TicketDTO:
    id: int
    code: str
    state: str
    entry_name: str
    entry_start_at: datetime
    entry_end_at: datetime | None = None
    ticket_type_name: str | None = None
    calendar_name: str | None = None
    created_at: datetime | None = None
    checked_in_at: datetime | None = None
    entry_id: int | None = None
    ticket_type_id: int | None = None
    price: Decimal | None = None
    order_id: int | None = None
    calendar_container_id: int | None = None


@dataclass(frozen=True, slots=True)
class CalendarEntryDTO:
    id: int
    calendar_id: int
    name: str
    start_at: datetime
    state: str
    cost: Decimal
    end_at: datetime | None = None
    user_id: int | None = None
    session_id: str | None = None
    order_id: int | None = None
    slot_id: int | None = None
    ticket_price: Decimal | None = None
    ticket_count: int | None = None
    calendar_name: str | None = None
    calendar_slug: str | None = None
    calendar_container_id: int | None = None
    calendar_container_type: str | None = None


# ---------------------------------------------------------------------------
# Market domain
# ---------------------------------------------------------------------------

@dataclass(frozen=True, slots=True)
class MarketPlaceDTO:
    id: int
    container_type: str
    container_id: int
    name: str
    slug: str
    description: str | None = None


@dataclass(frozen=True, slots=True)
class ProductDTO:
    id: int
    slug: str
    title: str | None = None
    image: str | None = None
    description_short: str | None = None
    rrp: Decimal | None = None
    regular_price: Decimal | None = None
    special_price: Decimal | None = None


# ---------------------------------------------------------------------------
# Cart domain
# ---------------------------------------------------------------------------

@dataclass(frozen=True, slots=True)
class CartItemDTO:
    id: int
    product_id: int
    quantity: int
    product_title: str | None = None
    product_slug: str | None = None
    product_image: str | None = None
    unit_price: Decimal | None = None
    market_place_id: int | None = None


@dataclass(frozen=True, slots=True)
class CartSummaryDTO:
    count: int = 0
    total: Decimal = Decimal("0")
    calendar_count: int = 0
    calendar_total: Decimal = Decimal("0")
    items: list[CartItemDTO] = field(default_factory=list)
    ticket_count: int = 0
    ticket_total: Decimal = Decimal("0")


# ---------------------------------------------------------------------------
# Federation / ActivityPub domain
# ---------------------------------------------------------------------------

@dataclass(frozen=True, slots=True)
class ActorProfileDTO:
    id: int
    user_id: int
    preferred_username: str
    public_key_pem: str
    display_name: str | None = None
    summary: str | None = None
    inbox_url: str | None = None
    outbox_url: str | None = None
    created_at: datetime | None = None


@dataclass(frozen=True, slots=True)
class APActivityDTO:
    id: int
    activity_id: str
    activity_type: str
    actor_profile_id: int
    object_type: str | None = None
    object_data: dict | None = None
    published: datetime | None = None
    is_local: bool = True
    source_type: str | None = None
    source_id: int | None = None
    ipfs_cid: str | None = None


@dataclass(frozen=True, slots=True)
class APFollowerDTO:
    id: int
    actor_profile_id: int
    follower_acct: str
    follower_inbox: str
    follower_actor_url: str
    created_at: datetime | None = None
    app_domain: str = "federation"


@dataclass(frozen=True, slots=True)
class APAnchorDTO:
    id: int
    merkle_root: str
    activity_count: int = 0
    tree_ipfs_cid: str | None = None
    ots_proof_cid: str | None = None
    confirmed_at: datetime | None = None
    bitcoin_txid: str | None = None


@dataclass(frozen=True, slots=True)
class RemoteActorDTO:
    id: int
    actor_url: str
    inbox_url: str
    preferred_username: str
    domain: str
    display_name: str | None = None
    summary: str | None = None
    icon_url: str | None = None
    shared_inbox_url: str | None = None
    public_key_pem: str | None = None


@dataclass(frozen=True, slots=True)
class RemotePostDTO:
    id: int
    remote_actor_id: int
    object_id: str
    content: str
    summary: str | None = None
    url: str | None = None
    attachments: list[dict] = field(default_factory=list)
    tags: list[dict] = field(default_factory=list)
    published: datetime | None = None
    actor: RemoteActorDTO | None = None


@dataclass(frozen=True, slots=True)
class TimelineItemDTO:
    id: str  # composite key for cursor pagination
    post_type: str  # "local" | "remote" | "boost"
    content: str  # HTML
    published: datetime
    actor_name: str
    actor_username: str
    object_id: str | None = None
    summary: str | None = None
    url: str | None = None
    attachments: list[dict] = field(default_factory=list)
    tags: list[dict] = field(default_factory=list)
    actor_domain: str | None = None  # None = local
    actor_icon: str | None = None
    actor_url: str | None = None
    boosted_by: str | None = None
    like_count: int = 0
    boost_count: int = 0
    liked_by_me: bool = False
    boosted_by_me: bool = False
    author_inbox: str | None = None


@dataclass(frozen=True, slots=True)
class NotificationDTO:
    id: int
    notification_type: str  # follow/like/boost/mention/reply
    from_actor_name: str
    from_actor_username: str
    created_at: datetime
    read: bool
    from_actor_domain: str | None = None
    from_actor_icon: str | None = None
    target_content_preview: str | None = None
    app_domain: str | None = None