Upload app/services/account_service.py

app/services/account_service.py

@@ -0,0 +1,107 @@
+"""Account service: create, get, list (with filters), update, soft-delete."""
+
+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.schemas.account import AccountCreate, AccountUpdate
+
+
+class AccountNotFound(Exception):
+    """Raised when an account lookup fails."""
+
+
+async def create_account(
+    db: AsyncSession, payload: AccountCreate, *, org_id: int, owner_id: int
+) -> Account:
+    """Create a new account in the given org."""
+    from app.services._base import OrgScopedQuery
+
+    account = Account(
+        org_id=org_id,
+        name=payload.name,
+        website=payload.website,
+        industry=payload.industry,
+        size=payload.size,
+        address=payload.address,
+        owner_id=owner_id,
+    )
+    db.add(account)
+    await db.commit()
+    await db.refresh(account)
+    return account
+
+
+async def get_account(
+    db: AsyncSession, account_id: int, *, org_id: int
+) -> Optional[Account]:
+    """Fetch a single account by id (org-scoped, not soft-deleted)."""
+    from app.services._base import OrgScopedQuery
+
+    q = OrgScopedQuery(Account, db, org_id=org_id)
+    return await q.get(account_id)
+
+
+async def list_accounts(
+    db: AsyncSession,
+    *,
+    org_id: int,
+    skip: int = 0,
+    limit: int = 20,
+    industry: Optional[str] = None,
+    size: Optional[str] = None,
+    owner_id: Optional[int] = None,
+    q: Optional[str] = None,
+) -> list[Account]:
+    """List accounts with filters and search."""
+    from app.services._base import OrgScopedQuery
+
+    scoped = OrgScopedQuery(Account, db, org_id=org_id)
+    base = await scoped.list(
+        skip=skip,
+        limit=limit,
+        order_by=Account.id,
+        industry=industry,
+        size=size,
+        owner_id=owner_id,
+    )
+    if q:
+        # In-memory filter for name (small datasets, fine for v1)
+        needle = q.lower()
+        base = [a for a in base if needle in a.name.lower()]
+    return base
+
+
+async def update_account(
+    db: AsyncSession, account: Account, payload: AccountUpdate
+) -> Account:
+    """Apply partial updates to an account."""
+    data = payload.model_dump(exclude_unset=True)
+    for field, value in data.items():
+        if value is not None:
+            setattr(account, field, value)
+    await db.commit()
+    await db.refresh(account)
+    return account
+
+
+async def soft_delete_account(db: AsyncSession, account: Account) -> Account:
+    """Soft-delete an account by setting deleted_at to now."""
+    account.deleted_at = datetime.now(UTC)
+    await db.commit()
+    await db.refresh(account)
+    return account
+
+
+__all__ = [
+    "AccountNotFound",
+    "create_account",
+    "get_account",
+    "list_accounts",
+    "soft_delete_account",
+    "update_account",
+]