Upload app/services/note_service.py

2026-06-04 00:06:26 +00:00
parent 2b9bd76a78
commit 8b3febe1a7
1 changed files with 146 additions and 0 deletions
@@ -0,0 +1,146 @@
+"""Note service: polymorphic notes attached to accounts/contacts/deals (R-2 validation)."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Optional
+
+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.note import Note, NoteParentType
+from app.schemas.note import NoteCreate, NoteUpdate
+from app.services._base import OrgScopedQuery
+
+
+class NoteNotFound(Exception):
+    """Raised when a note lookup fails."""
+
+
+class InvalidParent(Exception):
+    """Raised when note parent_type + parent_id don't point to an existing entity.
+
+    Mitigates R-2 (polymorphic validation): because parent_id has no DB-level FK,
+    the service layer must verify the parent exists before creating a note.
+    """
+
+
+async def _validate_parent(
+    db: AsyncSession,
+    *,
+    parent_type: NoteParentType,
+    parent_id: int,
+    org_id: int,
+) -> None:
+    """Verify the (parent_type, parent_id) tuple points to an existing entity in org."""
+    if parent_type == NoteParentType.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 == NoteParentType.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 == NoteParentType.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_note(
+    db: AsyncSession,
+    payload: NoteCreate,
+    *,
+    org_id: int,
+    author_id: int,
+) -> Note:
+    """Create a new note. R-2: validates parent_type + parent_id exist."""
+    parent_type = (
+        payload.parent_type
+        if isinstance(payload.parent_type, NoteParentType)
+        else NoteParentType(payload.parent_type)
+    )
+    await _validate_parent(
+        db,
+        parent_type=parent_type,
+        parent_id=payload.parent_id,
+        org_id=org_id,
+    )
+    note = Note(
+        org_id=org_id,
+        body=payload.body,
+        author_id=author_id,
+        parent_type=parent_type,
+        parent_id=payload.parent_id,
+    )
+    db.add(note)
+    await db.commit()
+    await db.refresh(note)
+    return note
+
+
+async def get_note(
+    db: AsyncSession, note_id: int, *, org_id: int
+) -> Optional[Note]:
+    """Fetch a single note by id."""
+    q = OrgScopedQuery(Note, db, org_id=org_id)
+    return await q.get(note_id)
+
+
+async def list_notes(
+    db: AsyncSession,
+    *,
+    org_id: int,
+    skip: int = 0,
+    limit: int = 20,
+    parent_type: Optional[str] = None,
+    parent_id: Optional[int] = None,
+) -> list[Note]:
+    """List notes with optional parent filters."""
+    scoped = OrgScopedQuery(Note, db, org_id=org_id)
+    return await scoped.list(
+        skip=skip,
+        limit=limit,
+        order_by=Note.id,
+        parent_type=parent_type,
+        parent_id=parent_id,
+    )
+
+
+async def update_note(
+    db: AsyncSession, note: Note, payload: NoteUpdate
+) -> Note:
+    """Apply partial updates to a note. Does NOT change parent (notes are pinned)."""
+    data = payload.model_dump(exclude_unset=True)
+    data.pop("parent_type", None)
+    data.pop("parent_id", None)
+    for field, value in data.items():
+        if value is not None:
+            setattr(note, field, value)
+    await db.commit()
+    await db.refresh(note)
+    return note
+
+
+async def soft_delete_note(db: AsyncSession, note: Note) -> Note:
+    """Soft-delete a note."""
+    note.deleted_at = datetime.now(UTC)
+    await db.commit()
+    await db.refresh(note)
+    return note
+
+
+__all__ = [
+    "InvalidParent",
+    "NoteNotFound",
+    "create_note",
+    "get_note",
+    "list_notes",
+    "soft_delete_note",
+    "update_note",
+]