Upload app/services/contact_service.py

app/services/contact_service.py

@@ -0,0 +1,135 @@
+"""Contact service: create, get, list (with filters), update, soft-delete."""
+
+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.schemas.contact import ContactCreate, ContactUpdate
+from app.services._base import OrgScopedQuery
+
+
+class ContactNotFound(Exception):
+    """Raised when a contact lookup fails."""
+
+
+class InvalidAccount(Exception):
+    """Raised when account_id points to a non-existent account."""
+
+
+async def _validate_account(
+    db: AsyncSession, account_id: int, *, org_id: int
+) -> None:
+    """Raise if account_id does not exist within the org."""
+    if account_id is None:
+        return
+    q = OrgScopedQuery(Account, db, org_id=org_id)
+    exists = await q.get(account_id)
+    if exists is None:
+        raise InvalidAccount(f"Account {account_id} not found in this org")
+
+
+async def create_contact(
+    db: AsyncSession,
+    payload: ContactCreate,
+    *,
+    org_id: int,
+    owner_id: int,
+) -> Contact:
+    """Create a new contact. Validates account_id if provided."""
+    await _validate_account(db, payload.account_id, org_id=org_id)
+    contact = Contact(
+        org_id=org_id,
+        first_name=payload.first_name,
+        last_name=payload.last_name,
+        email=payload.email,
+        phone=payload.phone,
+        account_id=payload.account_id,
+        owner_id=owner_id,
+    )
+    db.add(contact)
+    try:
+        await db.commit()
+    except IntegrityError as e:
+        await db.rollback()
+        raise InvalidAccount(str(e)) from e
+    await db.refresh(contact)
+    return contact
+
+
+async def get_contact(
+    db: AsyncSession, contact_id: int, *, org_id: int
+) -> Optional[Contact]:
+    """Fetch a single contact by id."""
+    q = OrgScopedQuery(Contact, db, org_id=org_id)
+    return await q.get(contact_id)
+
+
+async def list_contacts(
+    db: AsyncSession,
+    *,
+    org_id: int,
+    skip: int = 0,
+    limit: int = 20,
+    account_id: Optional[int] = None,
+    owner_id: Optional[int] = None,
+    q: Optional[str] = None,
+) -> list[Contact]:
+    """List contacts with filters and search."""
+    scoped = OrgScopedQuery(Contact, db, org_id=org_id)
+    base = await scoped.list(
+        skip=skip,
+        limit=limit,
+        order_by=Contact.id,
+        account_id=account_id,
+        owner_id=owner_id,
+    )
+    if q:
+        needle = q.lower()
+        base = [
+            c
+            for c in base
+            if needle in c.first_name.lower()
+            or needle in c.last_name.lower()
+            or (c.email and needle in c.email.lower())
+        ]
+    return base
+
+
+async def update_contact(
+    db: AsyncSession, contact: Contact, payload: ContactUpdate
+) -> Contact:
+    """Apply partial updates to a contact."""
+    data = payload.model_dump(exclude_unset=True)
+    if "account_id" in data and data["account_id"] is not None:
+        await _validate_account(db, data["account_id"], org_id=contact.org_id)
+    for field, value in data.items():
+        if value is not None:
+            setattr(contact, field, value)
+    await db.commit()
+    await db.refresh(contact)
+    return contact
+
+
+async def soft_delete_contact(db: AsyncSession, contact: Contact) -> Contact:
+    """Soft-delete a contact."""
+    contact.deleted_at = datetime.now(UTC)
+    await db.commit()
+    await db.refresh(contact)
+    return contact
+
+
+__all__ = [
+    "ContactNotFound",
+    "InvalidAccount",
+    "create_contact",
+    "get_contact",
+    "list_contacts",
+    "soft_delete_contact",
+    "update_contact",
+]