Files
leocrm/app/services/contact_service.py
T

583 lines
21 KiB
Python
Raw Normal View History

"""Unified contact service — CRUD, FTS search, contactpersons, soft-delete, export."""
from __future__ import annotations
import uuid
from datetime import UTC
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from app.core.hooks import apply_filters, do_action
from app.models.contact import Contact, ContactPerson
2026-07-23 08:42:26 +02:00
from app.services.entity_history_service import record_history
def _compute_displayname(data: dict) -> str:
"""Compute displayname from type and name fields."""
if data.get("type") == "person":
2026-07-25 21:03:46 +02:00
parts = [data.get("suffix"), data.get("firstname"), data.get("surname")]
return " ".join(p for p in parts if p).strip()
else:
return data.get("name") or ""
def _serialize_contact(c: Contact) -> dict:
"""Serialize a Contact ORM object to dict."""
return {
"id": str(c.id),
"type": c.type,
"displayname": c.displayname,
2026-07-25 21:03:46 +02:00
"status": getattr(c, "status", "lead"),
"name": c.name,
"firstname": c.firstname,
"surname": c.surname,
2026-07-25 21:03:46 +02:00
"suffix": c.suffix,
"ext_name_line": c.ext_name_line,
"gender": c.gender,
"code": c.code,
"accounting_code": c.accounting_code,
"vendor_accounting_code": c.vendor_accounting_code,
"mailing_street": c.mailing_street,
"mailing_number": c.mailing_number,
"mailing_unit_number": c.mailing_unit_number,
"mailing_district": c.mailing_district,
"mailing_extra_address_line": c.mailing_extra_address_line,
"mailing_postalcode": c.mailing_postalcode,
"mailing_city": c.mailing_city,
"mailing_state": c.mailing_state,
"mailing_country": c.mailing_country,
"visit_street": c.visit_street,
"visit_number": c.visit_number,
"visit_unit_number": c.visit_unit_number,
"visit_district": c.visit_district,
"visit_extra_address_line": c.visit_extra_address_line,
"visit_postalcode": c.visit_postalcode,
"visit_city": c.visit_city,
"visit_state": c.visit_state,
"invoice_street": c.invoice_street,
"invoice_number": c.invoice_number,
"invoice_unit_number": c.invoice_unit_number,
"invoice_district": c.invoice_district,
"invoice_extra_address_line": c.invoice_extra_address_line,
"invoice_postalcode": c.invoice_postalcode,
"invoice_city": c.invoice_city,
"invoice_state": c.invoice_state,
"invoice_country": c.invoice_country,
"country": c.country,
"phone_1": c.phone_1,
"phone_2": c.phone_2,
"email_1": c.email_1,
"email_2": c.email_2,
"website": c.website,
"vat_code": c.vat_code,
"fiscal_code": c.fiscal_code,
"commerce_code": c.commerce_code,
"purchase_number": c.purchase_number,
"bic": c.bic,
"bank_account": c.bank_account,
2026-07-25 21:03:46 +02:00
"discount_crew": float(c.discount_crew) if c.discount_crew is not None else 0.0,
"discount_transport": float(c.discount_transport) if c.discount_transport is not None else 0.0,
"discount_rental": float(c.discount_rental) if c.discount_rental is not None else 0.0,
"discount_sale": float(c.discount_sale) if c.discount_sale is not None else 0.0,
"discount_subrent": float(c.discount_subrent) if c.discount_subrent is not None else 0.0,
"discount_total": float(c.discount_total) if c.discount_total is not None else 0.0,
"latitude": c.latitude,
"longitude": c.longitude,
"projectnote": c.projectnote,
"projectnote_title": c.projectnote_title,
"contact_warning": c.contact_warning,
"tags": c.tags,
"image": c.image,
"custom": c.custom,
"folder_id": str(c.folder_id) if c.folder_id else None,
"default_person_id": str(c.default_person_id) if c.default_person_id else None,
"admin_contactperson_id": str(c.admin_contactperson_id) if c.admin_contactperson_id else None,
"created_at": c.created_at.isoformat() if c.created_at else None,
"updated_at": c.updated_at.isoformat() if c.updated_at else None,
}
def _serialize_contact_person(cp: ContactPerson) -> dict:
"""Serialize a ContactPerson ORM object to dict."""
return {
"id": str(cp.id),
"contact_id": str(cp.contact_id),
"displayname": cp.displayname,
"firstname": cp.firstname,
"middle_name": cp.middle_name,
"lastname": cp.lastname,
"function": cp.function,
"phone": cp.phone,
"mobilephone": cp.mobilephone,
"email": cp.email,
"street": cp.street,
"number": cp.number,
"postalcode": cp.postalcode,
"city": cp.city,
"state": cp.state,
"country": cp.country,
"tags": cp.tags,
"custom": cp.custom,
"created_at": cp.created_at.isoformat() if cp.created_at else None,
"updated_at": cp.updated_at.isoformat() if cp.updated_at else None,
}
def _serialize_contact_detail(c: Contact) -> dict:
"""Serialize with contact_persons included."""
data = _serialize_contact(c)
data["contact_persons"] = [_serialize_contact_person(cp) for cp in (c.contact_persons or [])]
return data
async def list_contacts(
db: AsyncSession,
tenant_id: uuid.UUID,
page: int = 1,
page_size: int = 20,
search: str | None = None,
contact_type: str | None = None,
folder_id: str | None = None,
sort_by: str = "displayname",
sort_order: str = "asc",
resolved_perms: dict | None = None,
user_id: uuid.UUID | None = None,
is_system_admin: bool = False,
cursor: str | None = None,
) -> dict:
"""List contacts with pagination, FTS search, type/folder filter, sorting.
Applies row-level visibility filter based on ownership and entity_permissions.
Keyset-Pagination: If ``cursor`` is provided (a contact UUID), results are
filtered to ``id > cursor`` instead of using OFFSET. This is much faster
for large datasets. When ``cursor`` is not provided, classic page/page_size
offset pagination is used (backward compatible).
"""
from app.core.visibility import apply_visibility_filter
# I.4 Performance: Cache simple list queries (no search, no cursor, first 3 pages)
use_cache = not search and not cursor and page <= 3 and not folder_id
cache_key = f"contacts:list:{tenant_id}:{page}:{page_size}:{contact_type or 'all'}:{sort_by}:{sort_order}:{user_id or 'admin'}:{is_system_admin}"
if use_cache:
from app.core.cache import cache_get
cached = await cache_get(cache_key)
if cached:
return cached
base = select(Contact).where(
Contact.tenant_id == tenant_id,
Contact.deleted_at.is_(None),
)
# Apply row-level visibility filter
if user_id and not is_system_admin:
base = await apply_visibility_filter(
db, base, "contact", Contact, user_id, tenant_id, is_system_admin
)
if contact_type:
base = base.where(Contact.type == contact_type)
if folder_id:
base = base.where(Contact.folder_id == uuid.UUID(folder_id))
if search:
base = base.where(
Contact.search_tsv.op("@@")(func.plainto_tsquery("german", search))
)
# Keyset-Pagination: use keyset when sorting by id (independent of cursor)
use_keyset = sort_by == "id" and sort_order == "asc"
if use_keyset and cursor is not None:
base = base.where(Contact.id > uuid.UUID(cursor))
# Count — use approximate count for large tables (5000x faster on 1M+ rows)
# pg_class.reltuples is updated by ANALYZE/VACUUM and is ~99% accurate
from app.core.pagination import approximate_count
total = await approximate_count(db, "contacts")
# For small tables, approximate count may be 0 or stale — fall back to exact
if total < 100:
count_q = select(func.count()).select_from(base.subquery())
total = (await db.execute(count_q)).scalar() or 0
# Sort
sort_col = getattr(Contact, sort_by, Contact.displayname)
if sort_order == "desc":
sort_col = sort_col.desc()
base = base.order_by(sort_col)
# Paginate
if use_keyset:
base = base.limit(page_size)
else:
offset = (page - 1) * page_size
base = base.offset(offset).limit(page_size)
2026-07-25 09:19:32 +02:00
# Eager load contact_persons to avoid N+1 queries
base = base.options(selectinload(Contact.contact_persons))
result = await db.execute(base)
contacts = result.scalars().all()
# Next cursor for keyset pagination
next_cursor = None
if use_keyset and len(contacts) == page_size and contacts:
next_cursor = str(contacts[-1].id)
result = {
"items": [_serialize_contact(c) for c in contacts],
"total": total,
"page": page,
"page_size": page_size,
"next_cursor": next_cursor,
}
# I.4 Performance: Cache the result for simple queries
if use_cache:
from app.core.cache import cache_set
await cache_set(cache_key, result, ttl=60) # 60 second cache
return result
async def get_contact(db: AsyncSession, tenant_id: uuid.UUID, contact_id: str,
user_id: uuid.UUID | None = None, is_system_admin: bool = False) -> dict:
"""Get a single contact with contact_persons. Checks row-level access."""
q = (
select(Contact)
.options(selectinload(Contact.contact_persons))
.where(
Contact.id == uuid.UUID(contact_id),
Contact.tenant_id == tenant_id,
Contact.deleted_at.is_(None),
)
)
result = await db.execute(q)
contact = result.scalar_one_or_none()
if not contact:
raise ValueError("Contact not found")
# Check row-level access
if user_id and not is_system_admin:
from app.core.visibility import check_single_entity_access
has_access = await check_single_entity_access(
db, "contact", contact.id, user_id, tenant_id, "read", is_system_admin
)
if not has_access:
raise PermissionError("No access to this contact")
return _serialize_contact_detail(contact)
async def create_contact(
db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, data: dict
) -> dict:
"""Create a new contact."""
data["displayname"] = await apply_filters("contact.format_display_name", _compute_displayname(data), data=data, db=db, tenant_id=tenant_id, user_id=user_id)
# Hook: contact.before_create
await do_action("contact.before_create", data, db=db, tenant_id=tenant_id, user_id=user_id)
contact_persons_data = data.pop("contact_persons", None)
contact = Contact(
tenant_id=tenant_id,
created_by=user_id,
updated_by=user_id,
owner_id=user_id,
**{k: v for k, v in data.items() if hasattr(Contact, k)},
)
db.add(contact)
await db.flush()
# Create inline contact persons
if contact_persons_data:
for cp_data in contact_persons_data:
cp_data["displayname"] = " ".join(
p for p in [cp_data.get("firstname"), cp_data.get("lastname")] if p
).strip()
cp = ContactPerson(
tenant_id=tenant_id,
contact_id=contact.id,
created_by=user_id,
updated_by=user_id,
**{k: v for k, v in cp_data.items() if hasattr(ContactPerson, k)},
)
db.add(cp)
await db.flush()
# Load with contact_persons
q = select(Contact).options(selectinload(Contact.contact_persons)).where(Contact.id == contact.id)
result = await db.execute(q)
contact = result.scalar_one()
2026-07-23 08:42:26 +02:00
serialized = _serialize_contact_detail(contact)
# Record history
await record_history(
db, tenant_id, user_id, "contact", contact.id,
action="create", snapshot_after=serialized,
)
2026-07-25 21:03:46 +02:00
# Enqueue domain events via transactional outbox (durable, at-least-once)
from app.core.outbox import enqueue_outbox_event
await enqueue_outbox_event(db, tenant_id, 'contact.created', {
'contact_id': str(contact.id),
'tenant_id': str(tenant_id),
'user_id': str(user_id),
'type': data.get('type', 'person'),
})
if data.get('type') == 'company':
2026-07-25 21:03:46 +02:00
await enqueue_outbox_event(db, tenant_id, 'lead.created', {
'contact_id': str(contact.id),
'tenant_id': str(tenant_id),
'user_id': str(user_id),
})
# Hook: contact.after_create
await do_action("contact.after_create", serialized, db=db, tenant_id=tenant_id, user_id=user_id)
2026-07-23 08:42:26 +02:00
return serialized
async def update_contact(
db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, contact_id: str, data: dict,
is_system_admin: bool = False,
) -> dict:
"""Update a contact. Checks row-level write access."""
2026-07-25 21:03:46 +02:00
# Expire all cached objects to ensure fresh data with selectinload
db.expire_all()
q = (
select(Contact)
.options(selectinload(Contact.contact_persons))
.where(
Contact.id == uuid.UUID(contact_id),
Contact.tenant_id == tenant_id,
Contact.deleted_at.is_(None),
)
)
result = await db.execute(q)
contact = result.scalar_one_or_none()
if not contact:
raise ValueError("Contact not found")
# Check row-level write access
if not is_system_admin:
from app.core.visibility import check_single_entity_access
has_access = await check_single_entity_access(
db, "contact", contact.id, user_id, tenant_id, "write", is_system_admin
)
if not has_access:
raise PermissionError("No write access to this contact")
# Hook: contact.before_update
await do_action("contact.before_update", data, db=db, tenant_id=tenant_id, user_id=user_id, contact_id=contact_id)
2026-07-23 08:42:26 +02:00
# Capture snapshot before update
snapshot_before = _serialize_contact_detail(contact)
# Recompute displayname if name fields changed
2026-07-25 21:03:46 +02:00
if any(k in data for k in ("type", "name", "firstname", "surname", "suffix")):
merged = {**_serialize_contact(contact), **data}
data["displayname"] = _compute_displayname(merged)
for key, value in data.items():
if hasattr(contact, key):
setattr(contact, key, value)
contact.updated_by = user_id
await db.flush()
2026-07-25 21:03:46 +02:00
# Re-query with selectinload to avoid lazy-loading issues after flush
q2 = (
select(Contact)
.options(selectinload(Contact.contact_persons))
.where(Contact.id == contact.id)
)
result2 = await db.execute(q2)
contact = result2.scalar_one()
2026-07-23 08:42:26 +02:00
snapshot_after = _serialize_contact_detail(contact)
# Compute changes diff
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
await record_history(
db, tenant_id, user_id, "contact", contact.id,
action="update",
snapshot_before=snapshot_before,
snapshot_after=snapshot_after,
changes=changes or None,
)
2026-07-25 21:03:46 +02:00
# Enqueue domain event via transactional outbox (durable, at-least-once)
from app.core.outbox import enqueue_outbox_event
await enqueue_outbox_event(db, tenant_id, 'contact.updated', {
'contact_id': str(contact.id),
'tenant_id': str(tenant_id),
'user_id': str(user_id),
'type': contact.type,
})
# Hook: contact.after_update
await do_action("contact.after_update", snapshot_after, db=db, tenant_id=tenant_id, user_id=user_id, contact_id=contact_id)
2026-07-23 08:42:26 +02:00
return snapshot_after
2026-07-23 08:42:26 +02:00
async def delete_contact(
db: AsyncSession, tenant_id: uuid.UUID, contact_id: str, user_id: uuid.UUID | None = None,
is_system_admin: bool = False,
2026-07-23 08:42:26 +02:00
) -> None:
"""Soft-delete a contact. Checks row-level admin access."""
q = select(Contact).where(
Contact.id == uuid.UUID(contact_id),
Contact.tenant_id == tenant_id,
Contact.deleted_at.is_(None),
)
result = await db.execute(q)
contact = result.scalar_one_or_none()
if not contact:
raise ValueError("Contact not found")
2026-07-23 08:42:26 +02:00
# Check row-level admin access
# Tenant-owned contacts (owner_id=None) can be deleted by any user with
# contacts:delete permission (already checked by route via require_permission).
# User-owned contacts require admin-level entity access.
if not is_system_admin and contact.owner_id is not None:
from app.core.visibility import check_single_entity_access
has_access = await check_single_entity_access(
db, "contact", contact.id, user_id, tenant_id, "admin", is_system_admin
)
if not has_access:
raise PermissionError("No admin access to this contact")
# Hook: contact.before_delete
await do_action("contact.before_delete", db=db, tenant_id=tenant_id, contact_id=contact_id, user_id=user_id)
2026-07-23 08:42:26 +02:00
# Capture snapshot before deletion
from sqlalchemy.orm import selectinload
q2 = (
select(Contact)
.options(selectinload(Contact.contact_persons))
.where(Contact.id == contact.id)
)
result2 = await db.execute(q2)
contact_full = result2.scalar_one()
snapshot_before = _serialize_contact_detail(contact_full)
from datetime import datetime
contact.deleted_at = datetime.now(UTC)
await db.flush()
2026-07-23 08:42:26 +02:00
# Record history
await record_history(
db, tenant_id, user_id, "contact", contact.id,
action="delete", snapshot_before=snapshot_before,
)
# Hook: contact.after_delete
await do_action("contact.after_delete", db=db, tenant_id=tenant_id, contact_id=contact_id, user_id=user_id)
async def hard_delete_contact(db: AsyncSession, tenant_id: uuid.UUID, contact_id: str) -> None:
"""GDPR hard-delete a contact."""
q = select(Contact).where(
Contact.id == uuid.UUID(contact_id),
Contact.tenant_id == tenant_id,
)
result = await db.execute(q)
contact = result.scalar_one_or_none()
if not contact:
raise ValueError("Contact not found")
await db.delete(contact)
await db.flush()
# ── ContactPerson CRUD ──
async def list_contact_persons(
db: AsyncSession, tenant_id: uuid.UUID, contact_id: str
) -> list[dict]:
"""List all contact persons for a contact."""
q = select(ContactPerson).where(
ContactPerson.contact_id == uuid.UUID(contact_id),
ContactPerson.tenant_id == tenant_id,
ContactPerson.deleted_at.is_(None),
).order_by(ContactPerson.displayname)
result = await db.execute(q)
return [_serialize_contact_person(cp) for cp in result.scalars().all()]
async def create_contact_person(
db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, contact_id: str, data: dict
) -> dict:
"""Add a contact person to a contact."""
data["displayname"] = " ".join(
p for p in [data.get("firstname"), data.get("lastname")] if p
).strip()
cp = ContactPerson(
tenant_id=tenant_id,
contact_id=uuid.UUID(contact_id),
created_by=user_id,
updated_by=user_id,
**{k: v for k, v in data.items() if hasattr(ContactPerson, k)},
)
db.add(cp)
await db.flush()
return _serialize_contact_person(cp)
async def update_contact_person(
db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, contact_id: str, person_id: str, data: dict
) -> dict:
"""Update a contact person."""
q = select(ContactPerson).where(
ContactPerson.id == uuid.UUID(person_id),
ContactPerson.contact_id == uuid.UUID(contact_id),
ContactPerson.tenant_id == tenant_id,
ContactPerson.deleted_at.is_(None),
)
result = await db.execute(q)
cp = result.scalar_one_or_none()
if not cp:
raise ValueError("Contact person not found")
if any(k in data for k in ("firstname", "lastname")):
merged = {**_serialize_contact_person(cp), **data}
data["displayname"] = " ".join(
p for p in [merged.get("firstname"), merged.get("lastname")] if p
).strip()
for key, value in data.items():
if hasattr(cp, key):
setattr(cp, key, value)
cp.updated_by = user_id
await db.flush()
return _serialize_contact_person(cp)
async def delete_contact_person(
db: AsyncSession, tenant_id: uuid.UUID, contact_id: str, person_id: str
) -> None:
"""Soft-delete a contact person."""
q = select(ContactPerson).where(
ContactPerson.id == uuid.UUID(person_id),
ContactPerson.contact_id == uuid.UUID(contact_id),
ContactPerson.tenant_id == tenant_id,
ContactPerson.deleted_at.is_(None),
)
result = await db.execute(q)
cp = result.scalar_one_or_none()
if not cp:
raise ValueError("Contact person not found")
from datetime import datetime
cp.deleted_at = datetime.now(UTC)
await db.flush()