refactor(backend/api): extract BannerConsent + BannerAdmin services (Step 4)
Phase 1 Step 4, file 2 of 18. Same cookbook as audit_routes (4a91814+883ef70) applied to banner_routes.py. compliance/api/banner_routes.py (653 LOC) is decomposed into: compliance/api/banner_routes.py (255) — thin handlers compliance/services/banner_consent_service.py (298) — public SDK surface compliance/services/banner_admin_service.py (238) — site/category/vendor CRUD compliance/services/_banner_serializers.py ( 81) — ORM-to-dict helpers shared between the two services compliance/schemas/banner.py ( 85) — Pydantic request models Split rationale: the SDK-facing endpoints (consent CRUD, config retrieval, export, stats) and the admin CRUD endpoints (sites + categories + vendors) have distinct audiences and different auth stories, and combined they would push the service file over the 500 hard cap. Two focused services is cleaner than one ~540-line god class. The shared ORM-to-dict helpers live in a private sibling module (_banner_serializers) rather than a static method on either service, so both services can import without a cycle. Handlers follow the established pattern: - Depends(get_consent_service) or Depends(get_admin_service) - `with translate_domain_errors():` wrapping the service call - Explicit return type annotations - ~3-5 lines per handler Services raise NotFoundError / ConflictError / ValidationError from compliance.domain; no HTTPException in the service layer. mypy.ini flips compliance.api.banner_routes from ignore_errors=True to False, joining audit_routes in the strict scope. The services carry the same scoped `# mypy: disable-error-code="arg-type,assignment"` header used by the audit services for the ORM Column[T] issue. Pydantic schemas moved to compliance.schemas.banner (mirroring the Step 3 schemas split). They were previously defined inline in banner_routes.py and not referenced by anything outside it, so no backwards-compat shim is needed. Verified: - 224/224 pytest (173 baseline + 26 audit integration + 25 banner integration) pass - tests/contracts/test_openapi_baseline.py green (360/484 unchanged) - mypy compliance/ -> Success: no issues found in 123 source files - All new files under the 300 soft target (largest: 298) - banner_routes.py drops from 653 -> 255 LOC (below hard cap) Hard-cap violations remaining: 16 (was 17). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Sharang Parnerkar
@@ -0,0 +1,81 @@
|
||||
"""
|
||||
Internal ORM-to-dict serializers shared by the banner services.
|
||||
|
||||
Kept as plain module-level functions (not part of either service class) so
|
||||
they can be imported by both ``banner_consent_service`` and
|
||||
``banner_admin_service`` without cyclic dependencies.
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
from compliance.db.banner_models import (
|
||||
BannerCategoryConfigDB,
|
||||
BannerConsentDB,
|
||||
BannerSiteConfigDB,
|
||||
BannerVendorConfigDB,
|
||||
)
|
||||
|
||||
|
||||
def consent_to_dict(c: BannerConsentDB) -> dict[str, Any]:
|
||||
return {
|
||||
"id": str(c.id),
|
||||
"site_id": c.site_id,
|
||||
"device_fingerprint": c.device_fingerprint,
|
||||
"categories": c.categories or [],
|
||||
"vendors": c.vendors or [],
|
||||
"ip_hash": c.ip_hash,
|
||||
"consent_string": c.consent_string,
|
||||
"expires_at": c.expires_at.isoformat() if c.expires_at 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 site_config_to_dict(s: BannerSiteConfigDB) -> dict[str, Any]:
|
||||
return {
|
||||
"id": str(s.id),
|
||||
"site_id": s.site_id,
|
||||
"site_name": s.site_name,
|
||||
"site_url": s.site_url,
|
||||
"banner_title": s.banner_title,
|
||||
"banner_description": s.banner_description,
|
||||
"privacy_url": s.privacy_url,
|
||||
"imprint_url": s.imprint_url,
|
||||
"dsb_name": s.dsb_name,
|
||||
"dsb_email": s.dsb_email,
|
||||
"theme": s.theme or {},
|
||||
"tcf_enabled": s.tcf_enabled,
|
||||
"is_active": s.is_active,
|
||||
"created_at": s.created_at.isoformat() if s.created_at else None,
|
||||
"updated_at": s.updated_at.isoformat() if s.updated_at else None,
|
||||
}
|
||||
|
||||
|
||||
def category_to_dict(c: BannerCategoryConfigDB) -> dict[str, Any]:
|
||||
return {
|
||||
"id": str(c.id),
|
||||
"site_config_id": str(c.site_config_id),
|
||||
"category_key": c.category_key,
|
||||
"name_de": c.name_de,
|
||||
"name_en": c.name_en,
|
||||
"description_de": c.description_de,
|
||||
"description_en": c.description_en,
|
||||
"is_required": c.is_required,
|
||||
"sort_order": c.sort_order,
|
||||
"is_active": c.is_active,
|
||||
}
|
||||
|
||||
|
||||
def vendor_to_dict(v: BannerVendorConfigDB) -> dict[str, Any]:
|
||||
return {
|
||||
"id": str(v.id),
|
||||
"site_config_id": str(v.site_config_id),
|
||||
"vendor_name": v.vendor_name,
|
||||
"vendor_url": v.vendor_url,
|
||||
"category_key": v.category_key,
|
||||
"description_de": v.description_de,
|
||||
"description_en": v.description_en,
|
||||
"cookie_names": v.cookie_names or [],
|
||||
"retention_days": v.retention_days,
|
||||
"is_active": v.is_active,
|
||||
}
|
||||
238
backend-compliance/compliance/services/banner_admin_service.py
Normal file
238
backend-compliance/compliance/services/banner_admin_service.py
Normal file
@@ -0,0 +1,238 @@
|
||||
# mypy: disable-error-code="arg-type,assignment"
|
||||
# SQLAlchemy 1.x Column() descriptors are Column[T] statically, T at runtime.
|
||||
"""
|
||||
Banner admin service — site config + category + vendor CRUD.
|
||||
|
||||
Phase 1 Step 4: extracted from ``compliance.api.banner_routes``.
|
||||
Covers the admin surface: site configs and the nested category and
|
||||
vendor collections.
|
||||
"""
|
||||
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from compliance.db.banner_models import (
|
||||
BannerCategoryConfigDB,
|
||||
BannerSiteConfigDB,
|
||||
BannerVendorConfigDB,
|
||||
)
|
||||
from compliance.domain import ConflictError, NotFoundError, ValidationError
|
||||
from compliance.schemas.banner import (
|
||||
CategoryConfigCreate,
|
||||
SiteConfigCreate,
|
||||
SiteConfigUpdate,
|
||||
VendorConfigCreate,
|
||||
)
|
||||
from compliance.services._banner_serializers import (
|
||||
category_to_dict,
|
||||
site_config_to_dict,
|
||||
vendor_to_dict,
|
||||
)
|
||||
|
||||
_UPDATABLE_SITE_FIELDS = (
|
||||
"site_name",
|
||||
"site_url",
|
||||
"banner_title",
|
||||
"banner_description",
|
||||
"privacy_url",
|
||||
"imprint_url",
|
||||
"dsb_name",
|
||||
"dsb_email",
|
||||
"theme",
|
||||
"tcf_enabled",
|
||||
"is_active",
|
||||
)
|
||||
|
||||
|
||||
class BannerAdminService:
|
||||
"""Business logic for the banner admin surface."""
|
||||
|
||||
def __init__(self, db: Session) -> None:
|
||||
self.db = db
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Internal lookups
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _site_or_raise(self, tenant_id: uuid.UUID, site_id: str) -> BannerSiteConfigDB:
|
||||
config = (
|
||||
self.db.query(BannerSiteConfigDB)
|
||||
.filter(
|
||||
BannerSiteConfigDB.tenant_id == tenant_id,
|
||||
BannerSiteConfigDB.site_id == site_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not config:
|
||||
raise NotFoundError("Site config not found")
|
||||
return config
|
||||
|
||||
@staticmethod
|
||||
def _parse_uuid(raw: str, label: str) -> uuid.UUID:
|
||||
try:
|
||||
return uuid.UUID(raw)
|
||||
except ValueError as exc:
|
||||
raise ValidationError(f"Invalid {label} ID") from exc
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Site configs
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def list_sites(self, tenant_id: str) -> list[dict[str, Any]]:
|
||||
tid = uuid.UUID(tenant_id)
|
||||
configs = (
|
||||
self.db.query(BannerSiteConfigDB)
|
||||
.filter(BannerSiteConfigDB.tenant_id == tid)
|
||||
.order_by(BannerSiteConfigDB.created_at.desc())
|
||||
.all()
|
||||
)
|
||||
return [site_config_to_dict(c) for c in configs]
|
||||
|
||||
def create_site(self, tenant_id: str, body: SiteConfigCreate) -> dict[str, Any]:
|
||||
tid = uuid.UUID(tenant_id)
|
||||
existing = (
|
||||
self.db.query(BannerSiteConfigDB)
|
||||
.filter(
|
||||
BannerSiteConfigDB.tenant_id == tid,
|
||||
BannerSiteConfigDB.site_id == body.site_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if existing:
|
||||
raise ConflictError(
|
||||
f"Site config for '{body.site_id}' already exists"
|
||||
)
|
||||
config = BannerSiteConfigDB(
|
||||
tenant_id=tid,
|
||||
site_id=body.site_id,
|
||||
site_name=body.site_name,
|
||||
site_url=body.site_url,
|
||||
banner_title=body.banner_title or "Cookie-Einstellungen",
|
||||
banner_description=body.banner_description,
|
||||
privacy_url=body.privacy_url,
|
||||
imprint_url=body.imprint_url,
|
||||
dsb_name=body.dsb_name,
|
||||
dsb_email=body.dsb_email,
|
||||
theme=body.theme or {},
|
||||
tcf_enabled=body.tcf_enabled,
|
||||
)
|
||||
self.db.add(config)
|
||||
self.db.commit()
|
||||
self.db.refresh(config)
|
||||
return site_config_to_dict(config)
|
||||
|
||||
def update_site(
|
||||
self, tenant_id: str, site_id: str, body: SiteConfigUpdate
|
||||
) -> dict[str, Any]:
|
||||
tid = uuid.UUID(tenant_id)
|
||||
config = self._site_or_raise(tid, site_id)
|
||||
for field in _UPDATABLE_SITE_FIELDS:
|
||||
val = getattr(body, field, None)
|
||||
if val is not None:
|
||||
setattr(config, field, val)
|
||||
config.updated_at = datetime.now(timezone.utc)
|
||||
self.db.commit()
|
||||
self.db.refresh(config)
|
||||
return site_config_to_dict(config)
|
||||
|
||||
def delete_site(self, tenant_id: str, site_id: str) -> None:
|
||||
tid = uuid.UUID(tenant_id)
|
||||
config = self._site_or_raise(tid, site_id)
|
||||
self.db.delete(config)
|
||||
self.db.commit()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Categories
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def list_categories(self, tenant_id: str, site_id: str) -> list[dict[str, Any]]:
|
||||
tid = uuid.UUID(tenant_id)
|
||||
config = self._site_or_raise(tid, site_id)
|
||||
cats = (
|
||||
self.db.query(BannerCategoryConfigDB)
|
||||
.filter(BannerCategoryConfigDB.site_config_id == config.id)
|
||||
.order_by(BannerCategoryConfigDB.sort_order)
|
||||
.all()
|
||||
)
|
||||
return [category_to_dict(c) for c in cats]
|
||||
|
||||
def create_category(
|
||||
self, tenant_id: str, site_id: str, body: CategoryConfigCreate
|
||||
) -> dict[str, Any]:
|
||||
tid = uuid.UUID(tenant_id)
|
||||
config = self._site_or_raise(tid, site_id)
|
||||
cat = BannerCategoryConfigDB(
|
||||
site_config_id=config.id,
|
||||
category_key=body.category_key,
|
||||
name_de=body.name_de,
|
||||
name_en=body.name_en,
|
||||
description_de=body.description_de,
|
||||
description_en=body.description_en,
|
||||
is_required=body.is_required,
|
||||
sort_order=body.sort_order,
|
||||
)
|
||||
self.db.add(cat)
|
||||
self.db.commit()
|
||||
self.db.refresh(cat)
|
||||
return category_to_dict(cat)
|
||||
|
||||
def delete_category(self, category_id: str) -> None:
|
||||
cid = self._parse_uuid(category_id, "category")
|
||||
cat = (
|
||||
self.db.query(BannerCategoryConfigDB)
|
||||
.filter(BannerCategoryConfigDB.id == cid)
|
||||
.first()
|
||||
)
|
||||
if not cat:
|
||||
raise NotFoundError("Category not found")
|
||||
self.db.delete(cat)
|
||||
self.db.commit()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Vendors
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def list_vendors(self, tenant_id: str, site_id: str) -> list[dict[str, Any]]:
|
||||
tid = uuid.UUID(tenant_id)
|
||||
config = self._site_or_raise(tid, site_id)
|
||||
vendors = (
|
||||
self.db.query(BannerVendorConfigDB)
|
||||
.filter(BannerVendorConfigDB.site_config_id == config.id)
|
||||
.all()
|
||||
)
|
||||
return [vendor_to_dict(v) for v in vendors]
|
||||
|
||||
def create_vendor(
|
||||
self, tenant_id: str, site_id: str, body: VendorConfigCreate
|
||||
) -> dict[str, Any]:
|
||||
tid = uuid.UUID(tenant_id)
|
||||
config = self._site_or_raise(tid, site_id)
|
||||
vendor = BannerVendorConfigDB(
|
||||
site_config_id=config.id,
|
||||
vendor_name=body.vendor_name,
|
||||
vendor_url=body.vendor_url,
|
||||
category_key=body.category_key,
|
||||
description_de=body.description_de,
|
||||
description_en=body.description_en,
|
||||
cookie_names=body.cookie_names,
|
||||
retention_days=body.retention_days,
|
||||
)
|
||||
self.db.add(vendor)
|
||||
self.db.commit()
|
||||
self.db.refresh(vendor)
|
||||
return vendor_to_dict(vendor)
|
||||
|
||||
def delete_vendor(self, vendor_id: str) -> None:
|
||||
vid = self._parse_uuid(vendor_id, "vendor")
|
||||
vendor = (
|
||||
self.db.query(BannerVendorConfigDB)
|
||||
.filter(BannerVendorConfigDB.id == vid)
|
||||
.first()
|
||||
)
|
||||
if not vendor:
|
||||
raise NotFoundError("Vendor not found")
|
||||
self.db.delete(vendor)
|
||||
self.db.commit()
|
||||
298
backend-compliance/compliance/services/banner_consent_service.py
Normal file
298
backend-compliance/compliance/services/banner_consent_service.py
Normal file
@@ -0,0 +1,298 @@
|
||||
# mypy: disable-error-code="arg-type,assignment"
|
||||
# SQLAlchemy 1.x Column() descriptors are Column[T] statically, T at runtime.
|
||||
"""
|
||||
Banner consent service — SDK-facing endpoints.
|
||||
|
||||
Phase 1 Step 4: extracted from ``compliance.api.banner_routes``.
|
||||
Covers public device consent CRUD, site config retrieval (for banner
|
||||
display), export, and per-site consent statistics.
|
||||
|
||||
Admin-side site/category/vendor management lives in
|
||||
``compliance.services.banner_admin_service.BannerAdminService``.
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
import uuid
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Any, Optional
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from compliance.db.banner_models import (
|
||||
BannerCategoryConfigDB,
|
||||
BannerConsentAuditLogDB,
|
||||
BannerConsentDB,
|
||||
BannerSiteConfigDB,
|
||||
BannerVendorConfigDB,
|
||||
)
|
||||
from compliance.domain import NotFoundError, ValidationError
|
||||
from compliance.services._banner_serializers import (
|
||||
category_to_dict,
|
||||
consent_to_dict,
|
||||
site_config_to_dict,
|
||||
vendor_to_dict,
|
||||
)
|
||||
|
||||
|
||||
class BannerConsentService:
|
||||
"""Business logic for public SDK banner consent endpoints."""
|
||||
|
||||
def __init__(self, db: Session) -> None:
|
||||
self.db = db
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Internal helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@staticmethod
|
||||
def _hash_ip(ip: Optional[str]) -> Optional[str]:
|
||||
if not ip:
|
||||
return None
|
||||
return hashlib.sha256(ip.encode()).hexdigest()[:16]
|
||||
|
||||
def _log(
|
||||
self,
|
||||
tenant_id: uuid.UUID,
|
||||
consent_id: Any,
|
||||
action: str,
|
||||
site_id: str,
|
||||
device_fingerprint: Optional[str] = None,
|
||||
categories: Optional[list[str]] = None,
|
||||
ip_hash: Optional[str] = None,
|
||||
) -> None:
|
||||
entry = BannerConsentAuditLogDB(
|
||||
tenant_id=tenant_id,
|
||||
consent_id=consent_id,
|
||||
action=action,
|
||||
site_id=site_id,
|
||||
device_fingerprint=device_fingerprint,
|
||||
categories=categories or [],
|
||||
ip_hash=ip_hash,
|
||||
)
|
||||
self.db.add(entry)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Consent CRUD (public SDK)
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def record_consent(
|
||||
self,
|
||||
tenant_id: str,
|
||||
site_id: str,
|
||||
device_fingerprint: str,
|
||||
categories: list[str],
|
||||
vendors: list[str],
|
||||
ip_address: Optional[str],
|
||||
user_agent: Optional[str],
|
||||
consent_string: Optional[str],
|
||||
) -> dict[str, Any]:
|
||||
"""Upsert a device consent row for (tenant, site, device_fingerprint)."""
|
||||
tid = uuid.UUID(tenant_id)
|
||||
ip_hash = self._hash_ip(ip_address)
|
||||
now = datetime.now(timezone.utc)
|
||||
expires_at = now + timedelta(days=365)
|
||||
|
||||
existing = (
|
||||
self.db.query(BannerConsentDB)
|
||||
.filter(
|
||||
BannerConsentDB.tenant_id == tid,
|
||||
BannerConsentDB.site_id == site_id,
|
||||
BannerConsentDB.device_fingerprint == device_fingerprint,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
if existing:
|
||||
existing.categories = categories
|
||||
existing.vendors = vendors
|
||||
existing.ip_hash = ip_hash
|
||||
existing.user_agent = user_agent
|
||||
existing.consent_string = consent_string
|
||||
existing.expires_at = expires_at
|
||||
existing.updated_at = now
|
||||
self.db.flush()
|
||||
self._log(
|
||||
tid, existing.id, "consent_updated", site_id, device_fingerprint,
|
||||
categories, ip_hash,
|
||||
)
|
||||
self.db.commit()
|
||||
self.db.refresh(existing)
|
||||
return consent_to_dict(existing)
|
||||
|
||||
consent = BannerConsentDB(
|
||||
tenant_id=tid,
|
||||
site_id=site_id,
|
||||
device_fingerprint=device_fingerprint,
|
||||
categories=categories,
|
||||
vendors=vendors,
|
||||
ip_hash=ip_hash,
|
||||
user_agent=user_agent,
|
||||
consent_string=consent_string,
|
||||
expires_at=expires_at,
|
||||
)
|
||||
self.db.add(consent)
|
||||
self.db.flush()
|
||||
self._log(
|
||||
tid, consent.id, "consent_given", site_id, device_fingerprint,
|
||||
categories, ip_hash,
|
||||
)
|
||||
self.db.commit()
|
||||
self.db.refresh(consent)
|
||||
return consent_to_dict(consent)
|
||||
|
||||
def get_consent(
|
||||
self, tenant_id: str, site_id: str, device_fingerprint: str
|
||||
) -> dict[str, Any]:
|
||||
"""Return the consent envelope for a device, or has_consent=false."""
|
||||
tid = uuid.UUID(tenant_id)
|
||||
consent = (
|
||||
self.db.query(BannerConsentDB)
|
||||
.filter(
|
||||
BannerConsentDB.tenant_id == tid,
|
||||
BannerConsentDB.site_id == site_id,
|
||||
BannerConsentDB.device_fingerprint == device_fingerprint,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not consent:
|
||||
return {"has_consent": False, "consent": None}
|
||||
return {"has_consent": True, "consent": consent_to_dict(consent)}
|
||||
|
||||
def withdraw_consent(self, tenant_id: str, consent_id: str) -> dict[str, Any]:
|
||||
"""Delete a consent row + audit the withdrawal."""
|
||||
tid = uuid.UUID(tenant_id)
|
||||
try:
|
||||
cid = uuid.UUID(consent_id)
|
||||
except ValueError as exc:
|
||||
raise ValidationError("Invalid consent ID") from exc
|
||||
|
||||
consent = (
|
||||
self.db.query(BannerConsentDB)
|
||||
.filter(BannerConsentDB.id == cid, BannerConsentDB.tenant_id == tid)
|
||||
.first()
|
||||
)
|
||||
if not consent:
|
||||
raise NotFoundError("Consent not found")
|
||||
|
||||
self._log(
|
||||
tid, cid, "consent_withdrawn", consent.site_id, consent.device_fingerprint,
|
||||
)
|
||||
self.db.delete(consent)
|
||||
self.db.commit()
|
||||
return {"success": True, "message": "Consent withdrawn"}
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Site config retrieval (SDK embed)
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def get_site_config(self, tenant_id: str, site_id: str) -> dict[str, Any]:
|
||||
"""Load site config + active categories + active vendors for banner display."""
|
||||
tid = uuid.UUID(tenant_id)
|
||||
config = (
|
||||
self.db.query(BannerSiteConfigDB)
|
||||
.filter(
|
||||
BannerSiteConfigDB.tenant_id == tid,
|
||||
BannerSiteConfigDB.site_id == site_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not config:
|
||||
return {
|
||||
"site_id": site_id,
|
||||
"banner_title": "Cookie-Einstellungen",
|
||||
"banner_description": (
|
||||
"Wir verwenden Cookies, um Ihnen die bestmoegliche Erfahrung zu bieten."
|
||||
),
|
||||
"categories": [],
|
||||
"vendors": [],
|
||||
}
|
||||
|
||||
categories = (
|
||||
self.db.query(BannerCategoryConfigDB)
|
||||
.filter(
|
||||
BannerCategoryConfigDB.site_config_id == config.id,
|
||||
BannerCategoryConfigDB.is_active,
|
||||
)
|
||||
.order_by(BannerCategoryConfigDB.sort_order)
|
||||
.all()
|
||||
)
|
||||
vendors = (
|
||||
self.db.query(BannerVendorConfigDB)
|
||||
.filter(
|
||||
BannerVendorConfigDB.site_config_id == config.id,
|
||||
BannerVendorConfigDB.is_active,
|
||||
)
|
||||
.all()
|
||||
)
|
||||
result = site_config_to_dict(config)
|
||||
result["categories"] = [category_to_dict(c) for c in categories]
|
||||
result["vendors"] = [vendor_to_dict(v) for v in vendors]
|
||||
return result
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# DSGVO export + stats
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def export_consent(
|
||||
self, tenant_id: str, site_id: str, device_fingerprint: str
|
||||
) -> dict[str, Any]:
|
||||
"""DSGVO export of all consent + audit rows for a device."""
|
||||
tid = uuid.UUID(tenant_id)
|
||||
consents = (
|
||||
self.db.query(BannerConsentDB)
|
||||
.filter(
|
||||
BannerConsentDB.tenant_id == tid,
|
||||
BannerConsentDB.site_id == site_id,
|
||||
BannerConsentDB.device_fingerprint == device_fingerprint,
|
||||
)
|
||||
.all()
|
||||
)
|
||||
audit = (
|
||||
self.db.query(BannerConsentAuditLogDB)
|
||||
.filter(
|
||||
BannerConsentAuditLogDB.tenant_id == tid,
|
||||
BannerConsentAuditLogDB.site_id == site_id,
|
||||
BannerConsentAuditLogDB.device_fingerprint == device_fingerprint,
|
||||
)
|
||||
.order_by(BannerConsentAuditLogDB.created_at.desc())
|
||||
.all()
|
||||
)
|
||||
return {
|
||||
"device_fingerprint": device_fingerprint,
|
||||
"site_id": site_id,
|
||||
"consents": [consent_to_dict(c) for c in consents],
|
||||
"audit_trail": [
|
||||
{
|
||||
"id": str(a.id),
|
||||
"action": a.action,
|
||||
"categories": a.categories or [],
|
||||
"created_at": a.created_at.isoformat() if a.created_at else None,
|
||||
}
|
||||
for a in audit
|
||||
],
|
||||
}
|
||||
|
||||
def get_site_stats(self, tenant_id: str, site_id: str) -> dict[str, Any]:
|
||||
"""Compute consent count + per-category acceptance rates for a site."""
|
||||
tid = uuid.UUID(tenant_id)
|
||||
base = self.db.query(BannerConsentDB).filter(
|
||||
BannerConsentDB.tenant_id == tid,
|
||||
BannerConsentDB.site_id == site_id,
|
||||
)
|
||||
total = base.count()
|
||||
category_stats: dict[str, int] = {}
|
||||
for c in base.all():
|
||||
cats: list[str] = list(c.categories or [])
|
||||
for cat in cats:
|
||||
category_stats[cat] = category_stats.get(cat, 0) + 1
|
||||
return {
|
||||
"site_id": site_id,
|
||||
"total_consents": total,
|
||||
"category_acceptance": {
|
||||
cat: {
|
||||
"count": count,
|
||||
"rate": round(count / total * 100, 1) if total > 0 else 0,
|
||||
}
|
||||
for cat, count in category_stats.items()
|
||||
},
|
||||
}
|
||||
Reference in New Issue
Block a user