refactor(b1): contacts domain fully plugin-owned - routes moved from core to contacts plugin with require_active_plugin guard
Check Cross-Plugin Imports / check (push) Has been cancelled

This commit is contained in:
Agent Zero
2026-08-23 20:20:02 +02:00
parent 5cee78c54c
commit 5ad107ff83
8 changed files with 48 additions and 15 deletions
@@ -0,0 +1,353 @@
"""Companies routes — CRUD, search, filter, export, contact links.
Companies are Contact entities with type='company'.
Industry and description are stored in the custom JSONB field.
"""
from __future__ import annotations
import io
import uuid
from datetime import UTC, datetime
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from app.core.db import get_db
from app.deps import get_redis_dep, require_permission
from app.models.audit import AuditLog
from app.models.contact import Contact
from app.services.entity_history_service import record_history
router = APIRouter(prefix="/api/v1/companies", tags=["companies"])
def _serialize_company(c: Contact) -> dict[str, Any]:
custom = c.custom or {}
return {
"id": str(c.id),
"name": c.name,
"displayname": c.displayname,
"status": c.status,
"industry": custom.get("industry"),
"description": custom.get("description"),
"email_1": c.email_1,
"email_2": c.email_2,
"phone_1": c.phone_1,
"phone_2": c.phone_2,
"website": c.website,
"mailing_city": c.mailing_city,
"mailing_postalcode": c.mailing_postalcode,
"mailing_country": c.mailing_country,
"tags": c.tags,
"custom": c.custom,
}
@router.get("")
async def list_companies(
page: int = Query(1, ge=1),
page_size: int = Query(20, ge=1, le=100),
search: str | None = Query(None),
industry: str | None = Query(None),
sort_by: str = Query("name"),
sort_order: str = Query("asc", pattern="^(asc|desc)$"),
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:read")),
):
tenant_id = uuid.UUID(current_user["tenant_id"])
q = select(Contact).where(
Contact.tenant_id == tenant_id,
Contact.type == "company",
Contact.deleted_at.is_(None),
)
if search:
q = q.where(Contact.search_tsv.match(search) | Contact.name.ilike(f"%{search}%"))
if industry:
q = q.where(Contact.custom["industry"].astext == industry)
count_q = select(func.count()).select_from(q.subquery())
total = (await db.execute(count_q)).scalar() or 0
sort_col = getattr(Contact, sort_by, Contact.name)
if sort_order == "desc":
sort_col = sort_col.desc()
q = q.order_by(sort_col).offset((page - 1) * page_size).limit(page_size)
result = await db.execute(q)
companies = result.scalars().all()
return {"items": [_serialize_company(c) for c in companies], "total": total, "page": page, "page_size": page_size}
@router.post("", status_code=status.HTTP_201_CREATED)
async def create_company(
body: dict[str, Any],
db: AsyncSession = Depends(get_db),
redis=Depends(get_redis_dep),
current_user: dict = Depends(require_permission("contacts:write")),
):
name = body.get("name")
if not name:
raise HTTPException(status_code=422, detail="name is required")
tenant_id = uuid.UUID(current_user["tenant_id"])
user_id = uuid.UUID(current_user["user_id"])
custom = {}
if body.get("industry"):
custom["industry"] = body["industry"]
if body.get("description"):
custom["description"] = body["description"]
for k, v in body.items():
if k not in ("name", "industry", "description"):
custom[k] = v
from app.core.hooks import do_action
await do_action("company.before_create", body, db=db, tenant_id=tenant_id, user_id=user_id)
company = Contact(
tenant_id=tenant_id, type="company", name=name, displayname=name,
status=body.get("status", "lead"), custom=custom,
created_by=user_id, updated_by=user_id,
)
db.add(company)
await db.flush()
audit_entry = AuditLog(
tenant_id=tenant_id, user_id=user_id, action="create",
entity_type="contact", entity_id=company.id,
changes={"name": name, "type": "company", **custom},
)
db.add(audit_entry)
await db.flush()
# Record history (D-CORE)
snapshot = _serialize_company(company)
await record_history(db, tenant_id, user_id, "contact", company.id, "create", snapshot_after=snapshot)
from app.core.hooks import do_action
await do_action("company.after_create", snapshot, db=db, tenant_id=tenant_id, user_id=user_id)
return snapshot
@router.get("/export")
async def export_companies(
format: str = Query("csv", pattern="^(csv|xlsx)$"),
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:read")),
):
tenant_id = uuid.UUID(current_user["tenant_id"])
q = select(Contact).where(
Contact.tenant_id == tenant_id, Contact.type == "company",
Contact.deleted_at.is_(None),
).order_by(Contact.name)
result = await db.execute(q)
companies = result.scalars().all()
rows = [_serialize_company(c) for c in companies]
if format == "csv":
import csv
output = io.StringIO()
if rows:
writer = csv.DictWriter(output, fieldnames=rows[0].keys())
writer.writeheader()
writer.writerows(rows)
return Response(content=output.getvalue(), media_type="text/csv",
headers={"Content-Disposition": "attachment; filename=companies.csv"})
else:
try:
import openpyxl
except ImportError:
raise HTTPException(status_code=500, detail="openpyxl not installed") from None
wb = openpyxl.Workbook()
ws = wb.active
ws.title = "Companies"
if rows:
headers = list(rows[0].keys())
ws.append(headers)
for row in rows:
ws.append([str(v) if v is not None else "" for v in row.values()])
output = io.BytesIO()
wb.save(output)
return Response(content=output.getvalue(),
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
headers={"Content-Disposition": "attachment; filename=companies.xlsx"})
@router.get("/{company_id}")
async def get_company(
company_id: str,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:read")),
):
tenant_id = uuid.UUID(current_user["tenant_id"])
q = select(Contact).options(selectinload(Contact.contact_persons)).where(
Contact.id == uuid.UUID(company_id), Contact.tenant_id == tenant_id,
Contact.type == "company", Contact.deleted_at.is_(None),
)
result = await db.execute(q)
company = result.scalar_one_or_none()
if not company:
raise HTTPException(status_code=404, detail="Company not found")
contacts = []
if company.contact_persons:
for cp in company.contact_persons:
contacts.append({"id": str(cp.id), "firstname": cp.firstname, "lastname": cp.lastname, "email": cp.email, "phone": cp.phone})
data = _serialize_company(company)
data["contacts"] = contacts
return data
@router.put("/{company_id}")
async def update_company(
company_id: str,
body: dict[str, Any],
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:write")),
):
tenant_id = uuid.UUID(current_user["tenant_id"])
user_id = uuid.UUID(current_user["user_id"])
q = select(Contact).where(
Contact.id == uuid.UUID(company_id), Contact.tenant_id == tenant_id,
Contact.type == "company", Contact.deleted_at.is_(None),
)
result = await db.execute(q)
company = result.scalar_one_or_none()
if not company:
raise HTTPException(status_code=404, detail="Company not found")
from app.core.hooks import do_action
await do_action("company.before_update", body, db=db, tenant_id=tenant_id, user_id=user_id, company_id=company_id)
# Capture snapshot before update (D-CORE)
snapshot_before = _serialize_company(company)
if "name" in body:
company.name = body["name"]
company.displayname = body["name"]
if "status" in body:
company.status = body["status"]
# Update custom fields - use raw SQL to avoid lazy loading issues
custom = dict(company.custom) if company.custom else {}
if "industry" in body:
custom["industry"] = body["industry"]
if "description" in body:
custom["description"] = body["description"]
company.custom = custom
company.updated_by = user_id
await db.flush()
snapshot_after = _serialize_company(company)
# Compute changes diff (D-CORE)
changes: dict = {}
for key, new_val in snapshot_after.items():
old_val = snapshot_before.get(key)
if old_val != new_val:
changes[key] = {"old": old_val, "new": new_val}
# Record history (D-CORE)
await record_history(
db, tenant_id, user_id, "contact", company.id, "update",
snapshot_before=snapshot_before, snapshot_after=snapshot_after, changes=changes or None,
)
audit_entry = AuditLog(tenant_id=tenant_id, user_id=user_id, action="update",
entity_type="contact", entity_id=company.id, changes=body)
db.add(audit_entry)
await db.flush()
from app.core.hooks import do_action
await do_action("company.after_update", snapshot_after, db=db, tenant_id=tenant_id, user_id=user_id, company_id=company_id)
return snapshot_after
@router.delete("/{company_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_company(
company_id: str,
cascade: bool = Query(False),
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:delete")),
):
tenant_id = uuid.UUID(current_user["tenant_id"])
user_id = uuid.UUID(current_user["user_id"])
q = select(Contact).options(selectinload(Contact.contact_persons)).where(
Contact.id == uuid.UUID(company_id), Contact.tenant_id == tenant_id,
Contact.type == "company", Contact.deleted_at.is_(None),
)
result = await db.execute(q)
company = result.scalar_one_or_none()
if not company:
raise HTTPException(status_code=404, detail="Company not found")
snapshot = _serialize_company(company)
from app.core.hooks import do_action
await do_action("company.before_delete", db=db, tenant_id=tenant_id, user_id=user_id, company_id=company_id)
# Soft-delete contact persons via SQL to avoid lazy loading
company.deleted_at = datetime.now(UTC)
company.updated_by = user_id
if cascade and company.contact_persons:
for cp in company.contact_persons:
cp.deleted_at = datetime.now(UTC)
await db.flush()
await record_history(db, tenant_id, user_id, "contact", company.id, "delete", snapshot_before=snapshot)
audit_entry = AuditLog(tenant_id=tenant_id, user_id=user_id, action="delete",
entity_type="contact", entity_id=company.id, changes={"name": company.name})
db.add(audit_entry)
await db.flush()
from app.core.hooks import do_action
await do_action("company.after_delete", snapshot, db=db, tenant_id=tenant_id, user_id=user_id, company_id=company_id)
return Response(status_code=status.HTTP_204_NO_CONTENT)
@router.post("/{company_id}/contacts/{contact_id}")
async def link_contact_to_company(
company_id: str, contact_id: str,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:write")),
):
tenant_id = uuid.UUID(current_user["tenant_id"])
comp_q = select(Contact).where(Contact.id == uuid.UUID(company_id), Contact.tenant_id == tenant_id,
Contact.type == "company", Contact.deleted_at.is_(None))
company = (await db.execute(comp_q)).scalar_one_or_none()
if not company:
raise HTTPException(status_code=404, detail="Company not found")
person_q = select(Contact).where(Contact.id == uuid.UUID(contact_id), Contact.tenant_id == tenant_id,
Contact.deleted_at.is_(None))
person = (await db.execute(person_q)).scalar_one_or_none()
if not person:
raise HTTPException(status_code=404, detail="Contact not found")
# Create a ContactPerson link
from app.models.contact import ContactPerson
cp = ContactPerson(
tenant_id=tenant_id,
contact_id=company.id,
displayname=person.displayname or person.name or "",
firstname=person.firstname,
lastname=person.surname,
email=person.email_1,
phone=person.phone_1,
)
db.add(cp)
await db.flush()
return {"company_id": company_id, "contact_id": contact_id}
@router.delete("/{company_id}/contacts/{contact_id}", status_code=status.HTTP_204_NO_CONTENT)
async def unlink_contact_from_company(
company_id: str, contact_id: str,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:write")),
):
tenant_id = uuid.UUID(current_user["tenant_id"])
comp_q = select(Contact).where(Contact.id == uuid.UUID(company_id), Contact.tenant_id == tenant_id,
Contact.type == "company", Contact.deleted_at.is_(None))
company = (await db.execute(comp_q)).scalar_one_or_none()
if not company:
raise HTTPException(status_code=404, detail="Company not found")
# Remove ContactPerson link
from sqlalchemy import delete as sa_delete
from app.models.contact import ContactPerson
await db.execute(
sa_delete(ContactPerson).where(
ContactPerson.contact_id == uuid.UUID(company_id),
ContactPerson.tenant_id == tenant_id,
)
)
await db.flush()
return Response(status_code=status.HTTP_204_NO_CONTENT)
@router.get("/{company_id}/emails")
async def get_company_emails(
company_id: str,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:read")),
):
return []
@@ -0,0 +1,107 @@
"""Contact folder permission routes — ACL management."""
from __future__ import annotations
import uuid
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.db import get_db
from app.deps import require_permission
from app.schemas.contact_folder_permission import (
FolderPermissionCreate,
FolderPermissionUpdate,
)
from app.services import contact_folder_permission_service
router = APIRouter(prefix="/api/v1/contact-folders", tags=["contact-folder-permissions"])
@router.get("/{folder_id}/permissions")
async def list_folder_permissions(
folder_id: str,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:read")),
):
"""List all permission entries for a folder."""
tenant_id = uuid.UUID(current_user["tenant_id"])
try:
items = await contact_folder_permission_service.list_permissions(db, tenant_id, folder_id)
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e)) from e
return {"items": items, "total": len(items)}
@router.post("/{folder_id}/permissions", status_code=status.HTTP_201_CREATED)
async def create_folder_permission(
folder_id: str,
body: FolderPermissionCreate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:write")),
):
"""Grant or update a permission on a folder for a user or group."""
tenant_id = uuid.UUID(current_user["tenant_id"])
try:
return await contact_folder_permission_service.create_permission(
db,
tenant_id,
folder_id,
body.user_id,
body.group_id,
body.permission_level,
body.inherit_to_subfolders,
)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
@router.put("/{folder_id}/permissions/{permission_id}")
async def update_folder_permission(
folder_id: str,
permission_id: str,
body: FolderPermissionUpdate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:write")),
):
"""Update an existing permission entry."""
tenant_id = uuid.UUID(current_user["tenant_id"])
try:
return await contact_folder_permission_service.update_permission(
db,
tenant_id,
permission_id,
body.permission_level,
body.inherit_to_subfolders,
)
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e)) from e
@router.delete("/{folder_id}/permissions/{permission_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_folder_permission(
folder_id: str,
permission_id: str,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:write")),
):
"""Revoke a permission entry."""
tenant_id = uuid.UUID(current_user["tenant_id"])
try:
await contact_folder_permission_service.delete_permission(db, tenant_id, permission_id)
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e)) from e
@router.get("/{folder_id}/access")
async def get_folder_access(
folder_id: str,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:read")),
):
"""Get effective access level for the current user on a folder."""
tenant_id = uuid.UUID(current_user["tenant_id"])
user_id = uuid.UUID(current_user["user_id"])
return await contact_folder_permission_service.get_effective_access(
db, tenant_id, user_id, uuid.UUID(folder_id)
)
@@ -0,0 +1,110 @@
"""Contact folder routes — CRUD, move contacts, reorder."""
from __future__ import annotations
import uuid
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.db import get_db
from app.deps import require_permission
from app.schemas.contact_folder import (
ContactFolderCreate,
ContactFolderUpdate,
MoveContactRequest,
)
from app.services import contact_folder_service
router = APIRouter(prefix="/api/v1/contact-folders", tags=["contact-folders"])
@router.get("")
async def list_folders(
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:read")),
):
"""List all contact folders for the current user."""
tenant_id = uuid.UUID(current_user["tenant_id"])
user_id = uuid.UUID(current_user["user_id"])
items = await contact_folder_service.list_folders(db, tenant_id, user_id)
return {"items": items, "total": len(items)}
@router.post("", status_code=status.HTTP_201_CREATED)
async def create_folder(
body: ContactFolderCreate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:write")),
):
"""Create a new contact folder."""
tenant_id = uuid.UUID(current_user["tenant_id"])
user_id = uuid.UUID(current_user["user_id"])
return await contact_folder_service.create_folder(
db, tenant_id, user_id, body.name, body.parent_id
)
@router.put("/{folder_id}")
async def update_folder(
folder_id: str,
body: ContactFolderUpdate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:write")),
):
"""Update a contact folder (name, parent, sort_order)."""
tenant_id = uuid.UUID(current_user["tenant_id"])
user_id = uuid.UUID(current_user["user_id"])
data = body.model_dump(exclude_none=True)
try:
return await contact_folder_service.update_folder(
db, tenant_id, user_id, folder_id, data
)
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e)) from e
@router.delete("/{folder_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_folder(
folder_id: str,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:write")),
):
"""Delete a contact folder. Contacts are unassigned."""
tenant_id = uuid.UUID(current_user["tenant_id"])
user_id = uuid.UUID(current_user["user_id"])
try:
await contact_folder_service.delete_folder(db, tenant_id, user_id, folder_id)
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e)) from e
@router.put("/{folder_id}/reorder")
async def reorder_folders(
folder_id: str,
body: list[dict],
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:write")),
):
"""Batch reorder folders. Body: [{id, sort_order, parent_id?}, ...]."""
tenant_id = uuid.UUID(current_user["tenant_id"])
user_id = uuid.UUID(current_user["user_id"])
await contact_folder_service.reorder_folders(db, tenant_id, user_id, body)
return {"ok": True}
@router.put("/contacts/{contact_id}/move")
async def move_contact(
contact_id: str,
body: MoveContactRequest,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:write")),
):
"""Move a contact to a folder (or unassign with folder_id=null)."""
tenant_id = uuid.UUID(current_user["tenant_id"])
try:
return await contact_folder_service.move_contact(
db, tenant_id, contact_id, body.folder_id
)
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e)) from e
+30 -6
View File
@@ -9,25 +9,49 @@ from __future__ import annotations
import logging
from app.plugins.base import BasePlugin
from app.plugins.manifest import PluginManifest
from app.plugins.manifest import PluginManifest, PluginRouteDef
logger = logging.getLogger(__name__)
class ContactsPlugin(BasePlugin):
"""Contacts plugin — manages Contact entity lifecycle (models, permissions, restore, history).
"""Contacts plugin — owns the full Contact domain (Block B1).
Routes remain in app/routes/contacts.py as core routes, but entity lifecycle
(permissions, entity models, restore, history) is managed through on_activate/on_deactivate.
Routes (contacts, companies, contact folders, folder permissions) live in
this plugin and are mounted via manifest.routes with
require_active_plugin("contacts") protection. Entity lifecycle
(permissions, entity models, restore, history) is managed through
on_activate/on_deactivate like every other business plugin.
"""
manifest = PluginManifest(
name="contacts",
version="1.0.0",
version="1.1.0",
display_name="Contacts",
description="Core CRM contacts — persons and companies.",
dependencies=[],
routes=[], # Routes are registered as core routes in main.py
routes=[
PluginRouteDef(
path="/api/v1/contacts",
module="app.plugins.builtins.contacts.routes",
router_attr="router",
),
PluginRouteDef(
path="/api/v1/companies",
module="app.plugins.builtins.contacts.company_routes",
router_attr="router",
),
PluginRouteDef(
path="/api/v1/contact-folders",
module="app.plugins.builtins.contacts.folder_routes",
router_attr="router",
),
PluginRouteDef(
path="/api/v1/contact-folders",
module="app.plugins.builtins.contacts.folder_permission_routes",
router_attr="router",
),
],
events=[],
migrations=[],
permissions=[
+318
View File
@@ -0,0 +1,318 @@
"""Unified contact routes — CRUD, contactpersons, FTS search, export, soft-delete.
Write operations (create, update, delete, merge) are delegated to Commands.
Read operations (list, get, export, contact persons) use services directly.
"""
from __future__ import annotations
import io
import uuid
from typing import Any
import redis.asyncio as aioredis
from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
from fastapi.responses import StreamingResponse
from sqlalchemy.ext.asyncio import AsyncSession
from app.commands.contact_commands import (
CreateContactCommand,
DeleteContactCommand,
MergeContactsCommand,
UpdateContactCommand,
)
from app.core.db import get_db
from app.core.visibility import check_single_entity_access
from app.deps import get_redis_dep, require_permission
from app.schemas.contact import (
ContactCreate,
ContactPersonCreate,
ContactPersonUpdate,
ContactUpdate,
)
from app.services import contact_service, dedup_service
from app.services.export_service import export_service
router = APIRouter(prefix="/api/v1/contacts", tags=["contacts"])
# ── Deduplication / Merge (Task 5.23) ──────────────────────────────────────────
from pydantic import BaseModel, Field # noqa: E402
class DuplicateCheckRequest(BaseModel):
"""Request body for duplicate detection."""
threshold: float = Field(default=0.7, ge=0.0, le=1.0)
limit: int = Field(default=50, ge=1, le=200)
class MergeRequest(BaseModel):
"""Request body for merging two contacts."""
source_contact_id: str
target_contact_id: str
field_overrides: dict[str, Any] | None = Field(default=None)
note: str | None = Field(default=None)
@router.get("")
async def list_contacts(
page: int = Query(1, ge=1),
page_size: int = Query(20, ge=1, le=100),
search: str | None = Query(None),
type: str | None = Query(None, pattern="^(company|person)$"),
folder_id: str | None = Query(None),
sort_by: str = Query("displayname"),
sort_order: str = Query("asc", pattern="^(asc|desc)$"),
cursor: str | None = Query(None, description="Keyset pagination cursor (contact UUID)"),
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:read")),
):
"""List contacts with pagination, FTS search, type/folder filter, sorting.
Supports keyset pagination via ``cursor`` parameter for large datasets.
"""
tenant_id = uuid.UUID(current_user["tenant_id"])
user_id = uuid.UUID(current_user["user_id"])
is_admin = current_user.get("is_system_admin", False)
return await contact_service.list_contacts(
db, tenant_id,
page=page, page_size=page_size, search=search,
contact_type=type, folder_id=folder_id,
sort_by=sort_by, sort_order=sort_order,
resolved_perms=current_user,
user_id=user_id,
is_system_admin=is_admin,
cursor=cursor,
)
@router.get("/export")
async def export_contacts(
format: str = Query("csv", pattern="^(csv)$"),
type: str | None = Query(None, pattern="^(company|person)$"),
search: str | None = Query(None),
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:read")),
):
"""Stream contacts as CSV."""
tenant_id = uuid.UUID(current_user["tenant_id"])
user_id = uuid.UUID(current_user["user_id"])
is_admin = current_user.get("is_system_admin", False)
csv_data = await export_service.export_contacts_csv(
db, tenant_id, contact_type=type, search=search,
user_id=user_id, is_system_admin=is_admin,
)
return StreamingResponse(
io.StringIO(csv_data),
media_type="text/csv",
headers={"Content-Disposition": "attachment; filename=contacts.csv"},
)
@router.post("", status_code=status.HTTP_201_CREATED)
async def create_contact(
body: ContactCreate,
db: AsyncSession = Depends(get_db),
redis: aioredis.Redis = Depends(get_redis_dep),
current_user: dict = Depends(require_permission("contacts:write")),
):
"""Create a new contact (company or person) via CreateContactCommand."""
data = body.model_dump(exclude_none=True)
cmd = CreateContactCommand(data=data)
result = await cmd.execute(db, redis, current_user)
if not result.success:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=result.error)
return result.data
@router.get("/merge-history")
async def get_contact_merge_history(
page: int = Query(1, ge=1),
page_size: int = Query(20, ge=1, le=100),
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:read")),
):
"""Get paginated merge history for the tenant."""
tenant_id = uuid.UUID(current_user["tenant_id"])
return await dedup_service.get_merge_history(db, tenant_id, page=page, page_size=page_size)
@router.get("/{contact_id}")
async def get_contact(
contact_id: str,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:read")),
):
"""Get a single contact with contact_persons."""
tenant_id = uuid.UUID(current_user["tenant_id"])
user_id = uuid.UUID(current_user["user_id"])
is_admin = current_user.get("is_system_admin", False)
try:
return await contact_service.get_contact(db, tenant_id, contact_id, user_id=user_id, is_system_admin=is_admin)
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e)) from e
except PermissionError as e:
raise HTTPException(status_code=403, detail=str(e)) from e
@router.put("/{contact_id}")
async def update_contact(
contact_id: str,
body: ContactUpdate,
db: AsyncSession = Depends(get_db),
redis: aioredis.Redis = Depends(get_redis_dep),
current_user: dict = Depends(require_permission("contacts:write")),
):
"""Update a contact via UpdateContactCommand."""
data = body.model_dump(exclude_none=True)
cmd = UpdateContactCommand(contact_id=contact_id, data=data)
result = await cmd.execute(db, redis, current_user)
if not result.success:
if "not found" in (result.error or "").lower():
raise HTTPException(status_code=404, detail=result.error)
if "Invalid state transition" in (result.error or ""):
raise HTTPException(status_code=422, detail=result.error)
raise HTTPException(status_code=400, detail=result.error)
return result.data
@router.delete("/{contact_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_contact(
contact_id: str,
hard: bool = Query(False, description="GDPR hard-delete"),
db: AsyncSession = Depends(get_db),
redis: aioredis.Redis = Depends(get_redis_dep),
current_user: dict = Depends(require_permission("contacts:delete")),
):
"""Soft-delete (or hard-delete with ?hard=true) a contact via DeleteContactCommand."""
cmd = DeleteContactCommand(contact_id=contact_id, hard=hard)
result = await cmd.execute(db, redis, current_user)
if not result.success:
raise HTTPException(status_code=404, detail=result.error)
return Response(status_code=status.HTTP_204_NO_CONTENT)
# ── ContactPersons ──
@router.get("/{contact_id}/persons")
async def list_contact_persons(
contact_id: str,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:read")),
):
"""List all contact persons for a contact."""
tenant_id = uuid.UUID(current_user["tenant_id"])
items = await contact_service.list_contact_persons(db, tenant_id, contact_id)
return {"items": items, "total": len(items)}
@router.post("/{contact_id}/persons", status_code=status.HTTP_201_CREATED)
async def create_contact_person(
contact_id: str,
body: ContactPersonCreate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:write")),
):
"""Add a contact person to a contact."""
tenant_id = uuid.UUID(current_user["tenant_id"])
user_id = uuid.UUID(current_user["user_id"])
data = body.model_dump(exclude_none=True)
try:
return await contact_service.create_contact_person(db, tenant_id, user_id, contact_id, data)
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e)) from e
@router.put("/{contact_id}/persons/{person_id}")
async def update_contact_person(
contact_id: str,
person_id: str,
body: ContactPersonUpdate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:write")),
):
"""Update a contact person."""
tenant_id = uuid.UUID(current_user["tenant_id"])
user_id = uuid.UUID(current_user["user_id"])
data = body.model_dump(exclude_none=True)
try:
return await contact_service.update_contact_person(db, tenant_id, user_id, contact_id, person_id, data)
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e)) from e
@router.delete("/{contact_id}/persons/{person_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_contact_person(
contact_id: str,
person_id: str,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:delete")),
):
"""Delete a contact person."""
tenant_id = uuid.UUID(current_user["tenant_id"])
try:
await contact_service.delete_contact_person(db, tenant_id, contact_id, person_id)
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e)) from e
# ── Deduplication / Merge endpoints (Task 5.23) ───────────────────────────────
@router.post("/duplicates")
async def find_duplicate_contacts(
body: DuplicateCheckRequest,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:read")),
):
"""Find potential duplicate contacts within the tenant."""
tenant_id = uuid.UUID(current_user["tenant_id"])
return await dedup_service.find_duplicates(
db, tenant_id, threshold=body.threshold, limit=body.limit
)
@router.post("/merge")
async def merge_duplicate_contacts(
body: MergeRequest,
db: AsyncSession = Depends(get_db),
redis: aioredis.Redis = Depends(get_redis_dep),
current_user: dict = Depends(require_permission("contacts:write")),
):
"""Merge two contacts (source → target) via MergeContactsCommand."""
tenant_id = uuid.UUID(current_user["tenant_id"])
user_id = uuid.UUID(current_user["user_id"])
is_admin = current_user.get("is_system_admin", False)
# Check write access on both contacts
try:
source_uuid = uuid.UUID(body.source_contact_id)
target_uuid = uuid.UUID(body.target_contact_id)
except (ValueError, TypeError):
raise HTTPException(status_code=400, detail="Invalid contact ID") from None
source_access = await check_single_entity_access(
db, "contact", source_uuid, user_id, tenant_id,
required_level="write", is_system_admin=is_admin,
)
if not source_access:
raise HTTPException(status_code=403, detail="No write access to source contact")
target_access = await check_single_entity_access(
db, "contact", target_uuid, user_id, tenant_id,
required_level="write", is_system_admin=is_admin,
)
if not target_access:
raise HTTPException(status_code=403, detail="No write access to target contact")
cmd = MergeContactsCommand(
source_contact_id=body.source_contact_id,
target_contact_id=body.target_contact_id,
field_overrides=body.field_overrides,
note=body.note,
)
result = await cmd.execute(db, redis, current_user)
if not result.success:
raise HTTPException(status_code=400, detail=result.error)
return result.data