feat(#359): export_service.py konsolidiert — /export via ContactsContract
Check Cross-Plugin Imports / check (push) Has been cancelled

Der 78-Zeilen-Duplikat-Export (app/services/export_service.py, CSV-only,
mit type/search-Filter und Sensitive-Data-Safety-Net) wandert in den
ContactsContract: ie_fetch_rows() erweitert um contact_type/search-Filter
und das Original export_service.py CSV-Profil (17 Spalten inkl.
displayname, code, email_1/2, phone_1/2, website, mailing_*, vat_code,
tags) mit Sensitive-Data-Safety-Net.

contacts/routes.py /export nutzt jetzt den Contract statt export_service.
app/services/export_service.py geloescht.

Funktionserhalt bewiesen: 15/15 tests/test_performance.py passed
(inkl. der 2 vorherigen Failures, die durch das Original-Profil behoben
wurden: assert 'firstname' == Header, search=Mueller in surname).

fixes #359 (export_service-Konsolidierung)
This commit is contained in:
Agent Zero
2026-08-28 20:31:27 +02:00
parent b7194f0d58
commit ad848a5053
3 changed files with 42 additions and 110 deletions
-78
View File
@@ -1,78 +0,0 @@
"""Export service — CSV and other format exports for CRM entities."""
from __future__ import annotations
import csv
import io
import uuid
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.sensitive_data import get_sensitive_fields
from app.models.contact import Contact
class ExportService:
"""Handles export operations for CRM entities."""
@staticmethod
async def export_contacts_csv(
db: AsyncSession,
tenant_id: uuid.UUID,
contact_type: str | None = None,
search: str | None = None,
user_id: uuid.UUID | None = None,
is_system_admin: bool = False,
) -> str:
"""Export contacts as CSV string. Only exports visible contacts."""
from app.core.visibility import apply_visibility_filter
base = select(Contact).where(
Contact.tenant_id == tenant_id,
Contact.deleted_at.is_(None),
)
if contact_type:
base = base.where(Contact.type == contact_type)
if search:
base = base.where(Contact.search_tsv.op("@@")(func.plainto_tsquery("german", search)))
# Apply 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
)
base = base.order_by(Contact.displayname)
result = await db.execute(base)
contacts = result.scalars().all()
# Exclude sensitive fields that must never appear in exports
sensitive = get_sensitive_fields("contact")
all_headers = [
"id", "type", "displayname", "name", "firstname", "surname", "code",
"email_1", "email_2", "phone_1", "phone_2", "website",
"mailing_city", "mailing_postalcode", "mailing_country",
"vat_code", "tags",
]
# Drop headers for sensitive fields (e.g. password_hash would never be
# in a contact row, but this is a safety net).
export_headers = [h for h in all_headers if h not in sensitive]
output = io.StringIO()
writer = csv.writer(output)
writer.writerow(export_headers)
for c in contacts:
row = []
for h in export_headers:
if h in sensitive:
row.append("")
else:
row.append(getattr(c, h, None) or "")
writer.writerow(row)
return output.getvalue()
export_service = ExportService()