147 lines
4.0 KiB
Python
147 lines
4.0 KiB
Python
"""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",
|
|
]
|