Upload app/services/tag_service.py

app/services/tag_service.py

@@ -0,0 +1,158 @@
+"""Tag service: tags and polymorphic tag_links (R-2 validation)."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Optional
+
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.account import Account
+from app.models.contact import Contact
+from app.models.deal import Deal
+from app.models.tag import Tag
+from app.models.tag_link import TagLink, TagLinkParentType
+from app.schemas.tag import TagCreate, TagLinkCreate
+from app.services._base import OrgScopedQuery
+
+
+class TagNotFound(Exception):
+    """Raised when a tag lookup fails."""
+
+
+class InvalidParent(Exception):
+    """Raised when tag_link parent_type + parent_id don't point to an existing entity."""
+
+
+class DuplicateTagLink(Exception):
+    """Raised when a tag is already linked to a parent (unique constraint)."""
+
+
+async def _validate_parent(
+    db: AsyncSession,
+    *,
+    parent_type: TagLinkParentType,
+    parent_id: int,
+    org_id: int,
+) -> None:
+    """Verify (parent_type, parent_id) points to an existing entity in org."""
+    if parent_type == TagLinkParentType.account:
+        if await OrgScopedQuery(Account, db, org_id=org_id).get(parent_id) is None:
+            raise InvalidParent(
+                f"Account {parent_id} not found in this org (R-2 validation)"
+            )
+    elif parent_type == TagLinkParentType.contact:
+        if await OrgScopedQuery(Contact, db, org_id=org_id).get(parent_id) is None:
+            raise InvalidParent(
+                f"Contact {parent_id} not found in this org (R-2 validation)"
+            )
+    elif parent_type == TagLinkParentType.deal:
+        if await OrgScopedQuery(Deal, db, org_id=org_id).get(parent_id) is None:
+            raise InvalidParent(
+                f"Deal {parent_id} not found in this org (R-2 validation)"
+            )
+
+
+async def create_tag(
+    db: AsyncSession, payload: TagCreate, *, org_id: int
+) -> Tag:
+    """Create a new tag in the given org."""
+    tag = Tag(
+        org_id=org_id,
+        name=payload.name,
+        color=payload.color,
+    )
+    db.add(tag)
+    await db.commit()
+    await db.refresh(tag)
+    return tag
+
+
+async def list_tags(db: AsyncSession, *, org_id: int) -> list[Tag]:
+    """List all tags in the org."""
+    scoped = OrgScopedQuery(Tag, db, org_id=org_id)
+    return await scoped.list(skip=0, limit=1000, order_by=Tag.id)
+
+
+async def get_tag(db: AsyncSession, tag_id: int, *, org_id: int) -> Optional[Tag]:
+    """Fetch a single tag by id."""
+    return await OrgScopedQuery(Tag, db, org_id=org_id).get(tag_id)
+
+
+async def link_tag(
+    db: AsyncSession, payload: TagLinkCreate, *, org_id: int
+) -> TagLink:
+    """Link a tag to an entity. R-2: validates parent exists."""
+    parent_type = (
+        payload.parent_type
+        if isinstance(payload.parent_type, TagLinkParentType)
+        else TagLinkParentType(payload.parent_type)
+    )
+    # Verify tag exists in this org
+    if await OrgScopedQuery(Tag, db, org_id=org_id).get(payload.tag_id) is None:
+        raise TagNotFound(f"Tag {payload.tag_id} not found in this org")
+    # Verify parent exists in this org (R-2)
+    await _validate_parent(
+        db,
+        parent_type=parent_type,
+        parent_id=payload.parent_id,
+        org_id=org_id,
+    )
+    link = TagLink(
+        org_id=org_id,
+        tag_id=payload.tag_id,
+        parent_type=parent_type,
+        parent_id=payload.parent_id,
+    )
+    db.add(link)
+    try:
+        await db.commit()
+    except IntegrityError as e:
+        await db.rollback()
+        raise DuplicateTagLink(
+            f"Tag {payload.tag_id} is already linked to {parent_type.value}:{payload.parent_id}"
+        ) from e
+    await db.refresh(link)
+    return link
+
+
+async def unlink_tag(
+    db: AsyncSession,
+    *,
+    tag_id: int,
+    parent_type: TagLinkParentType,
+    parent_id: int,
+    org_id: int,
+) -> bool:
+    """Unlink a tag from an entity. Returns True if a link was deleted."""
+    from sqlalchemy import select
+
+    parent_type_str = parent_type.value if hasattr(parent_type, "value") else str(parent_type)
+    result = await db.execute(
+        select(TagLink).where(
+            TagLink.org_id == org_id,
+            TagLink.tag_id == tag_id,
+            TagLink.parent_type == parent_type_str,
+            TagLink.parent_id == parent_id,
+            TagLink.deleted_at.is_(None),
+        )
+    )
+    link = result.scalar_one_or_none()
+    if link is None:
+        return False
+    link.deleted_at = datetime.now(UTC)
+    await db.commit()
+    return True
+
+
+__all__ = [
+    "DuplicateTagLink",
+    "InvalidParent",
+    "TagNotFound",
+    "create_tag",
+    "get_tag",
+    "link_tag",
+    "list_tags",
+    "unlink_tag",
+]