feat(phase-4b): backend business-logic, 8 entities, 32 endpoints, 58 tests
This commit is contained in:
CRM Bot
@@ -1 +1,8 @@
|
||||
"""Business logic service layer."""
|
||||
"""Service layer for the CRM system.
|
||||
|
||||
Submodules are imported directly by routers via
|
||||
`from app.services.<name> import <func>` to avoid circular-import issues
|
||||
during application startup.
|
||||
"""
|
||||
|
||||
__all__: list[str] = []
|
||||
|
||||
@@ -0,0 +1,79 @@
|
||||
"""OrgScopedQuery helper: central tenant-isolation layer (R-1 mitigation).
|
||||
|
||||
Every business query passes through this helper so that:
|
||||
- `org_id` is always applied (multi-tenant isolation)
|
||||
- `deleted_at IS NULL` is applied by default (soft-delete)
|
||||
- R-1 risk is mitigated: no query accidentally crosses tenant boundaries
|
||||
|
||||
In v1 (single-tenant) org_id defaults to 1. In v2 (multi-tenant) the
|
||||
router layer passes `current_user.org_id`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Optional
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
|
||||
class OrgScopedQuery:
|
||||
"""Zentraler Query-Helper, der jede Query mit org_id filtert.
|
||||
|
||||
v1: org_id = 1 (Single-Tenant-Default)
|
||||
v2: org_id = current_user.org_id
|
||||
"""
|
||||
|
||||
def __init__(self, model: type[Any], db: AsyncSession, org_id: int = 1) -> None:
|
||||
self.model = model
|
||||
self.db = db
|
||||
self.org_id = org_id
|
||||
|
||||
async def get(self, id: int) -> Any:
|
||||
"""Fetch a single record by id, scoped to org and not soft-deleted."""
|
||||
result = await self.db.execute(
|
||||
select(self.model).where(
|
||||
self.model.id == id,
|
||||
self.model.org_id == self.org_id,
|
||||
self.model.deleted_at.is_(None),
|
||||
)
|
||||
)
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
async def list(
|
||||
self,
|
||||
skip: int = 0,
|
||||
limit: int = 20,
|
||||
order_by: Optional[Any] = None,
|
||||
**filters: Any,
|
||||
) -> list[Any]:
|
||||
"""List records scoped to org, with optional filters and pagination."""
|
||||
stmt = select(self.model).where(
|
||||
self.model.org_id == self.org_id,
|
||||
self.model.deleted_at.is_(None),
|
||||
)
|
||||
for field, value in filters.items():
|
||||
if value is not None:
|
||||
stmt = stmt.where(getattr(self.model, field) == value)
|
||||
if order_by is not None:
|
||||
stmt = stmt.order_by(order_by)
|
||||
stmt = stmt.offset(skip).limit(limit)
|
||||
result = await self.db.execute(stmt)
|
||||
return list(result.scalars().all())
|
||||
|
||||
async def count(self, **filters: Any) -> int:
|
||||
"""Count records scoped to org, with optional filters."""
|
||||
from sqlalchemy import func as sa_func
|
||||
|
||||
stmt = select(sa_func.count()).select_from(self.model).where(
|
||||
self.model.org_id == self.org_id,
|
||||
self.model.deleted_at.is_(None),
|
||||
)
|
||||
for field, value in filters.items():
|
||||
if value is not None:
|
||||
stmt = stmt.where(getattr(self.model, field) == value)
|
||||
result = await self.db.execute(stmt)
|
||||
return int(result.scalar_one())
|
||||
|
||||
|
||||
__all__ = ["OrgScopedQuery"]
|
||||
@@ -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",
|
||||
]
|
||||
@@ -0,0 +1,184 @@
|
||||
"""Activity service: create, get, list, update, complete, soft-delete."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import UTC, datetime
|
||||
from typing import Optional
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.models.activity import Activity, ActivityType
|
||||
from app.schemas.activity import ActivityCompleteRequest, ActivityCreate, ActivityUpdate
|
||||
from app.services._base import OrgScopedQuery
|
||||
|
||||
|
||||
class ActivityNotFound(Exception):
|
||||
"""Raised when an activity lookup fails."""
|
||||
|
||||
|
||||
class NoParentException(Exception):
|
||||
"""Raised when an activity is created without any parent (account/contact/deal)."""
|
||||
|
||||
|
||||
async def _validate_parents(
|
||||
db: AsyncSession,
|
||||
*,
|
||||
org_id: int,
|
||||
account_id: Optional[int],
|
||||
contact_id: Optional[int],
|
||||
deal_id: Optional[int],
|
||||
) -> None:
|
||||
"""Ensure at least one parent is set and each provided id exists in org."""
|
||||
if account_id is None and contact_id is None and deal_id is None:
|
||||
raise NoParentException(
|
||||
"Activity must reference at least one of: account_id, contact_id, deal_id"
|
||||
)
|
||||
if account_id is not None:
|
||||
from app.models.account import Account
|
||||
if await OrgScopedQuery(Account, db, org_id=org_id).get(account_id) is None:
|
||||
raise NoParentException(f"Account {account_id} not found in this org")
|
||||
if contact_id is not None:
|
||||
from app.models.contact import Contact
|
||||
if await OrgScopedQuery(Contact, db, org_id=org_id).get(contact_id) is None:
|
||||
raise NoParentException(f"Contact {contact_id} not found in this org")
|
||||
if deal_id is not None:
|
||||
from app.models.deal import Deal
|
||||
if await OrgScopedQuery(Deal, db, org_id=org_id).get(deal_id) is None:
|
||||
raise NoParentException(f"Deal {deal_id} not found in this org")
|
||||
|
||||
|
||||
async def create_activity(
|
||||
db: AsyncSession,
|
||||
payload: ActivityCreate,
|
||||
*,
|
||||
org_id: int,
|
||||
owner_id: int,
|
||||
) -> Activity:
|
||||
"""Create a new activity. Validates parents (at least one required)."""
|
||||
await _validate_parents(
|
||||
db,
|
||||
org_id=org_id,
|
||||
account_id=payload.account_id,
|
||||
contact_id=payload.contact_id,
|
||||
deal_id=payload.deal_id,
|
||||
)
|
||||
activity = Activity(
|
||||
org_id=org_id,
|
||||
type=payload.type,
|
||||
subject=payload.subject,
|
||||
body=payload.body,
|
||||
due_date=payload.due_date,
|
||||
account_id=payload.account_id,
|
||||
contact_id=payload.contact_id,
|
||||
deal_id=payload.deal_id,
|
||||
owner_id=owner_id,
|
||||
)
|
||||
db.add(activity)
|
||||
await db.commit()
|
||||
await db.refresh(activity)
|
||||
return activity
|
||||
|
||||
|
||||
async def get_activity(
|
||||
db: AsyncSession, activity_id: int, *, org_id: int
|
||||
) -> Optional[Activity]:
|
||||
"""Fetch a single activity by id."""
|
||||
q = OrgScopedQuery(Activity, db, org_id=org_id)
|
||||
return await q.get(activity_id)
|
||||
|
||||
|
||||
async def list_activities(
|
||||
db: AsyncSession,
|
||||
*,
|
||||
org_id: int,
|
||||
skip: int = 0,
|
||||
limit: int = 20,
|
||||
type: Optional[ActivityType] = None,
|
||||
owner_id: Optional[int] = None,
|
||||
overdue: Optional[bool] = None,
|
||||
completed: Optional[bool] = None,
|
||||
) -> list[Activity]:
|
||||
"""List activities with optional filters."""
|
||||
scoped = OrgScopedQuery(Activity, db, org_id=org_id)
|
||||
base = await scoped.list(
|
||||
skip=skip,
|
||||
limit=limit,
|
||||
order_by=Activity.id,
|
||||
type=type.value if hasattr(type, "value") else type,
|
||||
owner_id=owner_id,
|
||||
)
|
||||
now = datetime.now(UTC).replace(tzinfo=None) # SQLite returns naive datetimes
|
||||
if overdue is True:
|
||||
base = [
|
||||
a for a in base
|
||||
if a.due_date is not None and a.due_date < now and a.completed_at is None
|
||||
]
|
||||
if completed is True:
|
||||
base = [a for a in base if a.completed_at is not None]
|
||||
elif completed is False:
|
||||
base = [a for a in base if a.completed_at is None]
|
||||
return base
|
||||
|
||||
|
||||
async def update_activity(
|
||||
db: AsyncSession, activity: Activity, payload: ActivityUpdate
|
||||
) -> Activity:
|
||||
"""Apply partial updates to an activity."""
|
||||
data = payload.model_dump(exclude_unset=True)
|
||||
# If parents change, re-validate
|
||||
if any(k in data for k in ("account_id", "contact_id", "deal_id")):
|
||||
new_acc = data.get("account_id", activity.account_id)
|
||||
new_con = data.get("contact_id", activity.contact_id)
|
||||
new_deal = data.get("deal_id", activity.deal_id)
|
||||
await _validate_parents(
|
||||
db,
|
||||
org_id=activity.org_id,
|
||||
account_id=new_acc,
|
||||
contact_id=new_con,
|
||||
deal_id=new_deal,
|
||||
)
|
||||
for field, value in data.items():
|
||||
if value is not None:
|
||||
setattr(activity, field, value)
|
||||
await db.commit()
|
||||
await db.refresh(activity)
|
||||
return activity
|
||||
|
||||
|
||||
async def complete_activity(
|
||||
db: AsyncSession,
|
||||
activity: Activity,
|
||||
payload: ActivityCompleteRequest,
|
||||
) -> Activity:
|
||||
"""Mark activity as completed; set completed_at and optionally outcome."""
|
||||
activity.completed_at = datetime.now(UTC)
|
||||
if payload.outcome is not None:
|
||||
# We store outcome in body (free-form text) for v1
|
||||
existing = activity.body or ""
|
||||
outcome_line = f"\n[Outcome] {payload.outcome}"
|
||||
activity.body = (existing + outcome_line).strip()
|
||||
await db.commit()
|
||||
await db.refresh(activity)
|
||||
return activity
|
||||
|
||||
|
||||
async def soft_delete_activity(
|
||||
db: AsyncSession, activity: Activity
|
||||
) -> Activity:
|
||||
"""Soft-delete an activity."""
|
||||
activity.deleted_at = datetime.now(UTC)
|
||||
await db.commit()
|
||||
await db.refresh(activity)
|
||||
return activity
|
||||
|
||||
|
||||
__all__ = [
|
||||
"ActivityNotFound",
|
||||
"NoParentException",
|
||||
"complete_activity",
|
||||
"create_activity",
|
||||
"get_activity",
|
||||
"list_activities",
|
||||
"soft_delete_activity",
|
||||
"update_activity",
|
||||
]
|
||||
@@ -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",
|
||||
]
|
||||
@@ -0,0 +1,66 @@
|
||||
"""Dashboard service: KPIs and activity feed."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import UTC, datetime
|
||||
from decimal import Decimal
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.models.activity import Activity
|
||||
from app.models.deal import Deal, DealStage
|
||||
from app.services._base import OrgScopedQuery
|
||||
|
||||
|
||||
async def get_kpis(db: AsyncSession, *, org_id: int) -> dict[str, object]:
|
||||
"""Compute headline KPIs for the org's dashboard."""
|
||||
scoped = OrgScopedQuery(Deal, db, org_id=org_id)
|
||||
open_stages = (DealStage.lead, DealStage.qualified, DealStage.proposal, DealStage.negotiation)
|
||||
|
||||
all_deals = await scoped.list(skip=0, limit=10_000, order_by=Deal.id)
|
||||
open_deals = [d for d in all_deals if d.stage in open_stages]
|
||||
open_deals_count = len(open_deals)
|
||||
|
||||
pipeline_value = float(
|
||||
sum((d.value for d in open_deals), Decimal("0"))
|
||||
)
|
||||
|
||||
now = datetime.now(UTC)
|
||||
# SQLite returns naive datetimes; strip tz for comparison
|
||||
month_start = now.replace(tzinfo=None, day=1, hour=0, minute=0, second=0, microsecond=0)
|
||||
won_this_month = [
|
||||
d for d in all_deals
|
||||
if d.stage == DealStage.won and d.created_at >= month_start
|
||||
]
|
||||
won_count = len(won_this_month)
|
||||
|
||||
won_total = sum(1 for d in all_deals if d.stage == DealStage.won)
|
||||
lost_total = sum(1 for d in all_deals if d.stage == DealStage.lost)
|
||||
if won_total + lost_total == 0:
|
||||
conversion_rate = 0.0
|
||||
else:
|
||||
conversion_rate = round((won_total / (won_total + lost_total)) * 100, 2)
|
||||
|
||||
return {
|
||||
"open_deals_count": open_deals_count,
|
||||
"pipeline_value": pipeline_value,
|
||||
"won_this_month": won_count,
|
||||
"conversion_rate": conversion_rate,
|
||||
}
|
||||
|
||||
|
||||
async def get_activity_feed(
|
||||
db: AsyncSession, *, org_id: int, limit: int = 20
|
||||
) -> list[Activity]:
|
||||
"""Return the most recent activities, ordered by created_at desc."""
|
||||
result = await db.execute(
|
||||
select(Activity)
|
||||
.where(Activity.org_id == org_id, Activity.deleted_at.is_(None))
|
||||
.order_by(Activity.created_at.desc())
|
||||
.limit(limit)
|
||||
)
|
||||
return list(result.scalars().all())
|
||||
|
||||
|
||||
__all__ = ["get_activity_feed", "get_kpis"]
|
||||
@@ -0,0 +1,193 @@
|
||||
"""Deal service: create, get, list, update, update_stage, get_pipeline, soft-delete."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import defaultdict
|
||||
from datetime import UTC, datetime
|
||||
from decimal import Decimal
|
||||
from typing import Optional
|
||||
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.models.account import Account
|
||||
from app.models.deal import Deal, DealStage
|
||||
from app.models.deal_stage_history import DealStageHistory
|
||||
from app.schemas.deal import DealCreate, DealUpdate
|
||||
from app.services._base import OrgScopedQuery
|
||||
|
||||
|
||||
class DealNotFound(Exception):
|
||||
"""Raised when a deal lookup fails."""
|
||||
|
||||
|
||||
class InvalidAccount(Exception):
|
||||
"""Raised when account_id points to a non-existent account."""
|
||||
|
||||
|
||||
def _stage_value(stage: DealStage | str) -> str:
|
||||
"""Normalize stage to its string value."""
|
||||
if hasattr(stage, "value"):
|
||||
return stage.value # type: ignore[union-attr]
|
||||
return str(stage)
|
||||
|
||||
|
||||
async def create_deal(
|
||||
db: AsyncSession,
|
||||
payload: DealCreate,
|
||||
*,
|
||||
org_id: int,
|
||||
owner_id: int,
|
||||
) -> Deal:
|
||||
"""Create a new deal. Validates account_id exists in org."""
|
||||
acc_q = OrgScopedQuery(Account, db, org_id=org_id)
|
||||
if await acc_q.get(payload.account_id) is None:
|
||||
raise InvalidAccount(f"Account {payload.account_id} not found in this org")
|
||||
deal = Deal(
|
||||
org_id=org_id,
|
||||
title=payload.title,
|
||||
value=payload.value,
|
||||
currency=payload.currency,
|
||||
stage=payload.stage,
|
||||
close_date=payload.close_date,
|
||||
account_id=payload.account_id,
|
||||
owner_id=owner_id,
|
||||
won_lost_reason=payload.won_lost_reason,
|
||||
)
|
||||
db.add(deal)
|
||||
try:
|
||||
await db.commit()
|
||||
except IntegrityError as e:
|
||||
await db.rollback()
|
||||
raise InvalidAccount(str(e)) from e
|
||||
await db.refresh(deal)
|
||||
# Initial stage history entry (from None → initial stage)
|
||||
history = DealStageHistory(
|
||||
org_id=org_id,
|
||||
deal_id=deal.id,
|
||||
from_stage=None,
|
||||
to_stage=deal.stage,
|
||||
changed_by=owner_id,
|
||||
)
|
||||
db.add(history)
|
||||
await db.commit()
|
||||
await db.refresh(deal)
|
||||
return deal
|
||||
|
||||
|
||||
async def get_deal(
|
||||
db: AsyncSession, deal_id: int, *, org_id: int
|
||||
) -> Optional[Deal]:
|
||||
"""Fetch a single deal by id."""
|
||||
q = OrgScopedQuery(Deal, db, org_id=org_id)
|
||||
return await q.get(deal_id)
|
||||
|
||||
|
||||
async def list_deals(
|
||||
db: AsyncSession,
|
||||
*,
|
||||
org_id: int,
|
||||
skip: int = 0,
|
||||
limit: int = 20,
|
||||
stage: Optional[str] = None,
|
||||
owner_id: Optional[int] = None,
|
||||
account_id: Optional[int] = None,
|
||||
) -> list[Deal]:
|
||||
"""List deals with optional filters."""
|
||||
scoped = OrgScopedQuery(Deal, db, org_id=org_id)
|
||||
return await scoped.list(
|
||||
skip=skip,
|
||||
limit=limit,
|
||||
order_by=Deal.id,
|
||||
stage=stage,
|
||||
owner_id=owner_id,
|
||||
account_id=account_id,
|
||||
)
|
||||
|
||||
|
||||
async def update_deal(
|
||||
db: AsyncSession, deal: Deal, payload: DealUpdate
|
||||
) -> Deal:
|
||||
"""Apply partial updates to a deal. Does NOT change stage (use update_stage)."""
|
||||
data = payload.model_dump(exclude_unset=True)
|
||||
# Disallow direct stage changes via update endpoint
|
||||
data.pop("stage", None)
|
||||
for field, value in data.items():
|
||||
if value is not None:
|
||||
setattr(deal, field, value)
|
||||
await db.commit()
|
||||
await db.refresh(deal)
|
||||
return deal
|
||||
|
||||
|
||||
async def update_stage(
|
||||
db: AsyncSession,
|
||||
deal: Deal,
|
||||
new_stage: DealStage,
|
||||
*,
|
||||
changed_by: int,
|
||||
reason: Optional[str] = None,
|
||||
) -> Deal:
|
||||
"""Change a deal's stage and append a DealStageHistory record."""
|
||||
from_stage_str = _stage_value(deal.stage)
|
||||
to_stage_str = _stage_value(new_stage)
|
||||
if from_stage_str == to_stage_str:
|
||||
return deal # no-op
|
||||
|
||||
deal.stage = new_stage # type: ignore[assignment]
|
||||
if new_stage in (DealStage.won, DealStage.lost) and reason is not None:
|
||||
deal.won_lost_reason = reason
|
||||
|
||||
history = DealStageHistory(
|
||||
org_id=deal.org_id,
|
||||
deal_id=deal.id,
|
||||
from_stage=from_stage_str,
|
||||
to_stage=to_stage_str,
|
||||
changed_by=changed_by,
|
||||
)
|
||||
db.add(history)
|
||||
await db.commit()
|
||||
await db.refresh(deal)
|
||||
return deal
|
||||
|
||||
|
||||
async def get_pipeline(
|
||||
db: AsyncSession, *, org_id: int
|
||||
) -> list[dict[str, object]]:
|
||||
"""Return deals grouped by stage for the pipeline view."""
|
||||
scoped = OrgScopedQuery(Deal, db, org_id=org_id)
|
||||
all_deals = await scoped.list(skip=0, limit=10_000, order_by=Deal.id)
|
||||
grouped: dict[str, list[Deal]] = defaultdict(list)
|
||||
for d in all_deals:
|
||||
grouped[_stage_value(d.stage)].append(d)
|
||||
return [
|
||||
{
|
||||
"stage": stage.value,
|
||||
"count": len(deals),
|
||||
"total_value": float(sum((d.value for d in deals), Decimal("0"))),
|
||||
"deals": deals,
|
||||
}
|
||||
for stage in DealStage
|
||||
for deals in [grouped.get(stage.value, [])]
|
||||
]
|
||||
|
||||
|
||||
async def soft_delete_deal(db: AsyncSession, deal: Deal) -> Deal:
|
||||
"""Soft-delete a deal."""
|
||||
deal.deleted_at = datetime.now(UTC)
|
||||
await db.commit()
|
||||
await db.refresh(deal)
|
||||
return deal
|
||||
|
||||
|
||||
__all__ = [
|
||||
"DealNotFound",
|
||||
"InvalidAccount",
|
||||
"create_deal",
|
||||
"get_deal",
|
||||
"get_pipeline",
|
||||
"list_deals",
|
||||
"soft_delete_deal",
|
||||
"update_deal",
|
||||
"update_stage",
|
||||
]
|
||||
@@ -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",
|
||||
]
|
||||
@@ -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",
|
||||
]
|
||||
Reference in New Issue
Block a user