rose-ash/shared/services/relations_impl.py

"""Service wrapper for relations module functions.

Wraps the module-level functions in shared.services.relationships into
a class so they can be called via the ``(service "relations" ...)`` primitive.
"""
from __future__ import annotations

from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession


def _serialize_rel(r) -> dict[str, Any]:
    return {
        "id": r.id,
        "parent_type": r.parent_type,
        "parent_id": r.parent_id,
        "child_type": r.child_type,
        "child_id": r.child_id,
        "sort_order": r.sort_order,
        "label": r.label,
        "relation_type": r.relation_type,
        "metadata": r.metadata_,
    }


class SqlRelationsService:

    async def get_children(
        self, session: AsyncSession, *,
        parent_type: str, parent_id: int,
        child_type: str | None = None,
        relation_type: str | None = None,
    ) -> list[dict[str, Any]]:
        from shared.services.relationships import get_children
        rels = await get_children(
            session, parent_type, parent_id, child_type,
            relation_type=relation_type,
        )
        return [_serialize_rel(r) for r in rels]

    async def get_parents(
        self, session: AsyncSession, *,
        child_type: str, child_id: int,
        parent_type: str | None = None,
        relation_type: str | None = None,
    ) -> list[dict[str, Any]]:
        from shared.services.relationships import get_parents
        rels = await get_parents(
            session, child_type, child_id, parent_type,
            relation_type=relation_type,
        )
        return [_serialize_rel(r) for r in rels]

    async def attach_child(
        self, session: AsyncSession, *,
        parent_type: str, parent_id: int,
        child_type: str, child_id: int,
        label: str | None = None,
        sort_order: int | None = None,
        relation_type: str | None = None,
        metadata: dict | None = None,
    ) -> dict[str, Any]:
        from shared.services.relationships import attach_child
        rel = await attach_child(
            session,
            parent_type=parent_type, parent_id=parent_id,
            child_type=child_type, child_id=child_id,
            label=label, sort_order=sort_order,
            relation_type=relation_type, metadata=metadata,
        )
        return _serialize_rel(rel)

    async def detach_child(
        self, session: AsyncSession, *,
        parent_type: str, parent_id: int,
        child_type: str, child_id: int,
        relation_type: str | None = None,
    ) -> bool:
        from shared.services.relationships import detach_child
        return await detach_child(
            session,
            parent_type=parent_type, parent_id=parent_id,
            child_type=child_type, child_id=child_id,
            relation_type=relation_type,
        )

    async def relate(
        self, session: AsyncSession, *,
        relation_type: str,
        from_id: int, to_id: int,
        label: str | None = None,
        sort_order: int | None = None,
        metadata: dict | None = None,
    ) -> dict[str, Any]:
        """Registry-aware relation creation with cardinality enforcement."""
        from shared.services.relationships import attach_child, get_children
        from shared.sx.relations import get_relation

        defn = get_relation(relation_type)
        if defn is None:
            raise ValueError(f"unknown relation_type: {relation_type}")

        if defn.cardinality == "one-to-one":
            existing = await get_children(
                session,
                parent_type=defn.from_type,
                parent_id=from_id,
                child_type=defn.to_type,
                relation_type=relation_type,
            )
            if existing:
                raise ValueError("one-to-one relation already exists")

        rel = await attach_child(
            session,
            parent_type=defn.from_type, parent_id=from_id,
            child_type=defn.to_type, child_id=to_id,
            label=label, sort_order=sort_order,
            relation_type=relation_type, metadata=metadata,
        )
        return _serialize_rel(rel)

    async def unrelate(
        self, session: AsyncSession, *,
        relation_type: str, from_id: int, to_id: int,
    ) -> bool:
        from shared.services.relationships import detach_child
        from shared.sx.relations import get_relation

        defn = get_relation(relation_type)
        if defn is None:
            raise ValueError(f"unknown relation_type: {relation_type}")

        return await detach_child(
            session,
            parent_type=defn.from_type, parent_id=from_id,
            child_type=defn.to_type, child_id=to_id,
            relation_type=relation_type,
        )

    async def can_relate(
        self, session: AsyncSession, *,
        relation_type: str, from_id: int,
    ) -> dict[str, Any]:
        from shared.services.relationships import get_children
        from shared.sx.relations import get_relation

        defn = get_relation(relation_type)
        if defn is None:
            return {"allowed": False, "reason": f"unknown relation_type: {relation_type}"}

        if defn.cardinality == "one-to-one":
            existing = await get_children(
                session,
                parent_type=defn.from_type,
                parent_id=from_id,
                child_type=defn.to_type,
                relation_type=relation_type,
            )
            if existing:
                return {"allowed": False, "reason": "one-to-one relation already exists"}

        return {"allowed": True}