From d3c26ea8968e2c8a346c3b2a98d063853ff43bf3 Mon Sep 17 00:00:00 2001
From: Leopoldadmin <leopold.hammerschmidt@media-on.de>
Date: Thu, 4 Jun 2026 00:06:26 +0000
Subject: [PATCH] Upload app/services/deal_service.py

---
 app/services/deal_service.py | 193 +++++++++++++++++++++++++++++++++++
 1 file changed, 193 insertions(+)
 create mode 100644 app/services/deal_service.py

diff --git a/app/services/deal_service.py b/app/services/deal_service.py
new file mode 100644
index 0000000..2e577d3
--- /dev/null
+++ b/app/services/deal_service.py
@@ -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",
+]