From 10073f3ef0754580e1074122c52003747d955020 Mon Sep 17 00:00:00 2001 From: Sharang Parnerkar <30073382+mighty840@users.noreply.github.com> Date: Tue, 7 Apr 2026 18:52:31 +0200 Subject: [PATCH] refactor(backend/api): extract BannerConsent + BannerAdmin services (Step 4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- .../compliance/api/banner_routes.py | 604 +++--------------- .../compliance/schemas/banner.py | 85 +++ .../services/_banner_serializers.py | 81 +++ .../services/banner_admin_service.py | 238 +++++++ .../services/banner_consent_service.py | 298 +++++++++ backend-compliance/mypy.ini | 4 +- .../tests/contracts/openapi.baseline.json | 252 +++++--- 7 files changed, 975 insertions(+), 587 deletions(-) create mode 100644 backend-compliance/compliance/schemas/banner.py create mode 100644 backend-compliance/compliance/services/_banner_serializers.py create mode 100644 backend-compliance/compliance/services/banner_admin_service.py create mode 100644 backend-compliance/compliance/services/banner_consent_service.py diff --git a/backend-compliance/compliance/api/banner_routes.py b/backend-compliance/compliance/api/banner_routes.py index 9acc2f4..a8c5eea 100644 --- a/backend-compliance/compliance/api/banner_routes.py +++ b/backend-compliance/compliance/api/banner_routes.py @@ -2,181 +2,50 @@ Banner Consent Routes — Device-basierte Cookie-Consents fuer Kunden-Websites. Public SDK-Endpoints (fuer Einbettung) + Admin-Endpoints (Konfiguration & Stats). + +Phase 1 Step 4 refactor: handlers are thin and delegate to +``BannerConsentService`` (SDK surface) or ``BannerAdminService`` (admin +CRUD). Domain errors raised by the services are translated to +HTTPException via ``translate_domain_errors``. Pydantic request schemas +live in ``compliance.schemas.banner``. """ -import uuid -import hashlib -from datetime import datetime, timedelta, timezone -from typing import Optional, List +from typing import Any, Optional -from fastapi import APIRouter, Depends, HTTPException, Query, Header -from pydantic import BaseModel +from fastapi import APIRouter, Depends, Header, Query from sqlalchemy.orm import Session from classroom_engine.database import get_db -from ..db.banner_models import ( - BannerConsentDB, BannerConsentAuditLogDB, - BannerSiteConfigDB, BannerCategoryConfigDB, BannerVendorConfigDB, +from compliance.api._http_errors import translate_domain_errors +from compliance.schemas.banner import ( + CategoryConfigCreate, + ConsentCreate, + SiteConfigCreate, + SiteConfigUpdate, + VendorConfigCreate, ) +from compliance.services.banner_admin_service import BannerAdminService +from compliance.services.banner_consent_service import BannerConsentService router = APIRouter(prefix="/banner", tags=["compliance-banner"]) DEFAULT_TENANT = "9282a473-5c95-4b3a-bf78-0ecc0ec71d3e" -# ============================================================================= -# Schemas -# ============================================================================= +# ---------------------------------------------------------------------- +# Dependencies +# ---------------------------------------------------------------------- -class ConsentCreate(BaseModel): - site_id: str - device_fingerprint: str - categories: List[str] = [] - vendors: List[str] = [] - ip_address: Optional[str] = None - user_agent: Optional[str] = None - consent_string: Optional[str] = None - - -class SiteConfigCreate(BaseModel): - site_id: str - site_name: Optional[str] = None - site_url: Optional[str] = None - banner_title: Optional[str] = None - banner_description: Optional[str] = None - privacy_url: Optional[str] = None - imprint_url: Optional[str] = None - dsb_name: Optional[str] = None - dsb_email: Optional[str] = None - theme: Optional[dict] = None - tcf_enabled: bool = False - - -class SiteConfigUpdate(BaseModel): - site_name: Optional[str] = None - site_url: Optional[str] = None - banner_title: Optional[str] = None - banner_description: Optional[str] = None - privacy_url: Optional[str] = None - imprint_url: Optional[str] = None - dsb_name: Optional[str] = None - dsb_email: Optional[str] = None - theme: Optional[dict] = None - tcf_enabled: Optional[bool] = None - is_active: Optional[bool] = None - - -class CategoryConfigCreate(BaseModel): - category_key: str - name_de: str - name_en: Optional[str] = None - description_de: Optional[str] = None - description_en: Optional[str] = None - is_required: bool = False - sort_order: int = 0 - - -class VendorConfigCreate(BaseModel): - vendor_name: str - vendor_url: Optional[str] = None - category_key: str - description_de: Optional[str] = None - description_en: Optional[str] = None - cookie_names: List[str] = [] - retention_days: int = 365 - - -# ============================================================================= -# Helpers -# ============================================================================= - -def _get_tenant(x_tenant_id: Optional[str] = Header(None, alias='X-Tenant-ID')) -> str: +def _get_tenant(x_tenant_id: Optional[str] = Header(None, alias="X-Tenant-ID")) -> str: return x_tenant_id or DEFAULT_TENANT -def _hash_ip(ip: Optional[str]) -> Optional[str]: - if not ip: - return None - return hashlib.sha256(ip.encode()).hexdigest()[:16] +def get_consent_service(db: Session = Depends(get_db)) -> BannerConsentService: + return BannerConsentService(db) -def _consent_to_dict(c: BannerConsentDB) -> dict: - 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: - 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: - 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: - 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, - } - - -def _log_banner_audit(db, tenant_id, consent_id, action, site_id, device_fingerprint=None, categories=None, ip_hash=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, - ) - db.add(entry) - return entry +def get_admin_service(db: Session = Depends(get_db)) -> BannerAdminService: + return BannerAdminService(db) # ============================================================================= @@ -187,58 +56,20 @@ def _log_banner_audit(db, tenant_id, consent_id, action, site_id, device_fingerp async def record_consent( body: ConsentCreate, tenant_id: str = Depends(_get_tenant), - db: Session = Depends(get_db), -): + service: BannerConsentService = Depends(get_consent_service), +) -> dict[str, Any]: """Record device consent (upsert by site_id + device_fingerprint).""" - tid = uuid.UUID(tenant_id) - ip_hash = _hash_ip(body.ip_address) - - # Upsert: check existing - existing = db.query(BannerConsentDB).filter( - BannerConsentDB.tenant_id == tid, - BannerConsentDB.site_id == body.site_id, - BannerConsentDB.device_fingerprint == body.device_fingerprint, - ).first() - - if existing: - existing.categories = body.categories - existing.vendors = body.vendors - existing.ip_hash = ip_hash - existing.user_agent = body.user_agent - existing.consent_string = body.consent_string - existing.expires_at = datetime.now(timezone.utc) + timedelta(days=365) - existing.updated_at = datetime.now(timezone.utc) - db.flush() - - _log_banner_audit( - db, tid, existing.id, "consent_updated", - body.site_id, body.device_fingerprint, body.categories, ip_hash, + with translate_domain_errors(): + return service.record_consent( + tenant_id=tenant_id, + site_id=body.site_id, + device_fingerprint=body.device_fingerprint, + categories=body.categories, + vendors=body.vendors, + ip_address=body.ip_address, + user_agent=body.user_agent, + consent_string=body.consent_string, ) - db.commit() - db.refresh(existing) - return _consent_to_dict(existing) - - consent = BannerConsentDB( - tenant_id=tid, - site_id=body.site_id, - device_fingerprint=body.device_fingerprint, - categories=body.categories, - vendors=body.vendors, - ip_hash=ip_hash, - user_agent=body.user_agent, - consent_string=body.consent_string, - expires_at=datetime.now(timezone.utc) + timedelta(days=365), - ) - db.add(consent) - db.flush() - - _log_banner_audit( - db, tid, consent.id, "consent_given", - body.site_id, body.device_fingerprint, body.categories, ip_hash, - ) - db.commit() - db.refresh(consent) - return _consent_to_dict(consent) @router.get("/consent") @@ -246,88 +77,33 @@ async def get_consent( site_id: str = Query(...), device_fingerprint: str = Query(...), tenant_id: str = Depends(_get_tenant), - db: Session = Depends(get_db), -): + service: BannerConsentService = Depends(get_consent_service), +) -> dict[str, Any]: """Retrieve consent for a device.""" - tid = uuid.UUID(tenant_id) - consent = 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)} + with translate_domain_errors(): + return service.get_consent(tenant_id, site_id, device_fingerprint) @router.delete("/consent/{consent_id}") async def withdraw_consent( consent_id: str, tenant_id: str = Depends(_get_tenant), - db: Session = Depends(get_db), -): + service: BannerConsentService = Depends(get_consent_service), +) -> dict[str, Any]: """Withdraw a banner consent.""" - tid = uuid.UUID(tenant_id) - try: - cid = uuid.UUID(consent_id) - except ValueError: - raise HTTPException(status_code=400, detail="Invalid consent ID") - - consent = db.query(BannerConsentDB).filter( - BannerConsentDB.id == cid, - BannerConsentDB.tenant_id == tid, - ).first() - if not consent: - raise HTTPException(status_code=404, detail="Consent not found") - - _log_banner_audit( - db, tid, cid, "consent_withdrawn", - consent.site_id, consent.device_fingerprint, - ) - - db.delete(consent) - db.commit() - return {"success": True, "message": "Consent withdrawn"} + with translate_domain_errors(): + return service.withdraw_consent(tenant_id, consent_id) @router.get("/config/{site_id}") async def get_site_config( site_id: str, tenant_id: str = Depends(_get_tenant), - db: Session = Depends(get_db), -): + service: BannerConsentService = Depends(get_consent_service), +) -> dict[str, Any]: """Load site configuration for banner display.""" - tid = uuid.UUID(tenant_id) - config = 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 = db.query(BannerCategoryConfigDB).filter( - BannerCategoryConfigDB.site_config_id == config.id, - BannerCategoryConfigDB.is_active, - ).order_by(BannerCategoryConfigDB.sort_order).all() - - vendors = 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 + with translate_domain_errors(): + return service.get_site_config(tenant_id, site_id) @router.get("/consent/export") @@ -335,122 +111,51 @@ async def export_consent( site_id: str = Query(...), device_fingerprint: str = Query(...), tenant_id: str = Depends(_get_tenant), - db: Session = Depends(get_db), -): + service: BannerConsentService = Depends(get_consent_service), +) -> dict[str, Any]: """DSGVO export of all consent data for a device.""" - tid = uuid.UUID(tenant_id) - - consents = db.query(BannerConsentDB).filter( - BannerConsentDB.tenant_id == tid, - BannerConsentDB.site_id == site_id, - BannerConsentDB.device_fingerprint == device_fingerprint, - ).all() - - audit = 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 - ], - } + with translate_domain_errors(): + return service.export_consent(tenant_id, site_id, device_fingerprint) # ============================================================================= -# Admin Endpoints +# Admin — Stats # ============================================================================= @router.get("/admin/stats/{site_id}") async def get_site_stats( site_id: str, tenant_id: str = Depends(_get_tenant), - db: Session = Depends(get_db), -): + service: BannerConsentService = Depends(get_consent_service), +) -> dict[str, Any]: """Consent statistics per site.""" - tid = uuid.UUID(tenant_id) - base = db.query(BannerConsentDB).filter( - BannerConsentDB.tenant_id == tid, - BannerConsentDB.site_id == site_id, - ) + with translate_domain_errors(): + return service.get_site_stats(tenant_id, site_id) - total = base.count() - - # Count category acceptance rates - category_stats = {} - all_consents = base.all() - for c in all_consents: - for cat in (c.categories or []): - 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() - }, - } +# ============================================================================= +# Admin — Sites +# ============================================================================= @router.get("/admin/sites") async def list_site_configs( tenant_id: str = Depends(_get_tenant), - db: Session = Depends(get_db), -): + service: BannerAdminService = Depends(get_admin_service), +) -> list[dict[str, Any]]: """List all site configurations.""" - tid = uuid.UUID(tenant_id) - configs = 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] + with translate_domain_errors(): + return service.list_sites(tenant_id) @router.post("/admin/sites") async def create_site_config( body: SiteConfigCreate, tenant_id: str = Depends(_get_tenant), - db: Session = Depends(get_db), -): + service: BannerAdminService = Depends(get_admin_service), +) -> dict[str, Any]: """Create a site configuration.""" - tid = uuid.UUID(tenant_id) - - existing = db.query(BannerSiteConfigDB).filter( - BannerSiteConfigDB.tenant_id == tid, - BannerSiteConfigDB.site_id == body.site_id, - ).first() - if existing: - raise HTTPException(status_code=409, detail=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, - ) - db.add(config) - db.commit() - db.refresh(config) - return _site_config_to_dict(config) + with translate_domain_errors(): + return service.create_site(tenant_id, body) @router.put("/admin/sites/{site_id}") @@ -458,72 +163,37 @@ async def update_site_config( site_id: str, body: SiteConfigUpdate, tenant_id: str = Depends(_get_tenant), - db: Session = Depends(get_db), -): + service: BannerAdminService = Depends(get_admin_service), +) -> dict[str, Any]: """Update a site configuration.""" - tid = uuid.UUID(tenant_id) - config = db.query(BannerSiteConfigDB).filter( - BannerSiteConfigDB.tenant_id == tid, - BannerSiteConfigDB.site_id == site_id, - ).first() - if not config: - raise HTTPException(status_code=404, detail="Site config not found") - - for field in ["site_name", "site_url", "banner_title", "banner_description", - "privacy_url", "imprint_url", "dsb_name", "dsb_email", - "theme", "tcf_enabled", "is_active"]: - val = getattr(body, field, None) - if val is not None: - setattr(config, field, val) - - config.updated_at = datetime.now(timezone.utc) - db.commit() - db.refresh(config) - return _site_config_to_dict(config) + with translate_domain_errors(): + return service.update_site(tenant_id, site_id, body) @router.delete("/admin/sites/{site_id}", status_code=204) async def delete_site_config( site_id: str, tenant_id: str = Depends(_get_tenant), - db: Session = Depends(get_db), -): + service: BannerAdminService = Depends(get_admin_service), +) -> None: """Delete a site configuration.""" - tid = uuid.UUID(tenant_id) - config = db.query(BannerSiteConfigDB).filter( - BannerSiteConfigDB.tenant_id == tid, - BannerSiteConfigDB.site_id == site_id, - ).first() - if not config: - raise HTTPException(status_code=404, detail="Site config not found") - - db.delete(config) - db.commit() + with translate_domain_errors(): + service.delete_site(tenant_id, site_id) # ============================================================================= -# Admin Category Endpoints +# Admin — Categories # ============================================================================= @router.get("/admin/sites/{site_id}/categories") async def list_categories( site_id: str, tenant_id: str = Depends(_get_tenant), - db: Session = Depends(get_db), -): + service: BannerAdminService = Depends(get_admin_service), +) -> list[dict[str, Any]]: """List categories for a site.""" - tid = uuid.UUID(tenant_id) - config = db.query(BannerSiteConfigDB).filter( - BannerSiteConfigDB.tenant_id == tid, - BannerSiteConfigDB.site_id == site_id, - ).first() - if not config: - raise HTTPException(status_code=404, detail="Site config not found") - - cats = 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] + with translate_domain_errors(): + return service.list_categories(tenant_id, site_id) @router.post("/admin/sites/{site_id}/categories") @@ -531,75 +201,36 @@ async def create_category( site_id: str, body: CategoryConfigCreate, tenant_id: str = Depends(_get_tenant), - db: Session = Depends(get_db), -): + service: BannerAdminService = Depends(get_admin_service), +) -> dict[str, Any]: """Create a category for a site.""" - tid = uuid.UUID(tenant_id) - config = db.query(BannerSiteConfigDB).filter( - BannerSiteConfigDB.tenant_id == tid, - BannerSiteConfigDB.site_id == site_id, - ).first() - if not config: - raise HTTPException(status_code=404, detail="Site config not found") - - 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, - ) - db.add(cat) - db.commit() - db.refresh(cat) - return _category_to_dict(cat) + with translate_domain_errors(): + return service.create_category(tenant_id, site_id, body) @router.delete("/admin/categories/{category_id}", status_code=204) async def delete_category( category_id: str, - db: Session = Depends(get_db), -): + service: BannerAdminService = Depends(get_admin_service), +) -> None: """Delete a category.""" - try: - cid = uuid.UUID(category_id) - except ValueError: - raise HTTPException(status_code=400, detail="Invalid category ID") - - cat = db.query(BannerCategoryConfigDB).filter(BannerCategoryConfigDB.id == cid).first() - if not cat: - raise HTTPException(status_code=404, detail="Category not found") - - db.delete(cat) - db.commit() + with translate_domain_errors(): + service.delete_category(category_id) # ============================================================================= -# Admin Vendor Endpoints +# Admin — Vendors # ============================================================================= @router.get("/admin/sites/{site_id}/vendors") async def list_vendors( site_id: str, tenant_id: str = Depends(_get_tenant), - db: Session = Depends(get_db), -): + service: BannerAdminService = Depends(get_admin_service), +) -> list[dict[str, Any]]: """List vendors for a site.""" - tid = uuid.UUID(tenant_id) - config = db.query(BannerSiteConfigDB).filter( - BannerSiteConfigDB.tenant_id == tid, - BannerSiteConfigDB.site_id == site_id, - ).first() - if not config: - raise HTTPException(status_code=404, detail="Site config not found") - - vendors = db.query(BannerVendorConfigDB).filter( - BannerVendorConfigDB.site_config_id == config.id, - ).all() - return [_vendor_to_dict(v) for v in vendors] + with translate_domain_errors(): + return service.list_vendors(tenant_id, site_id) @router.post("/admin/sites/{site_id}/vendors") @@ -607,47 +238,18 @@ async def create_vendor( site_id: str, body: VendorConfigCreate, tenant_id: str = Depends(_get_tenant), - db: Session = Depends(get_db), -): + service: BannerAdminService = Depends(get_admin_service), +) -> dict[str, Any]: """Create a vendor for a site.""" - tid = uuid.UUID(tenant_id) - config = db.query(BannerSiteConfigDB).filter( - BannerSiteConfigDB.tenant_id == tid, - BannerSiteConfigDB.site_id == site_id, - ).first() - if not config: - raise HTTPException(status_code=404, detail="Site config not found") - - 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, - ) - db.add(vendor) - db.commit() - db.refresh(vendor) - return _vendor_to_dict(vendor) + with translate_domain_errors(): + return service.create_vendor(tenant_id, site_id, body) @router.delete("/admin/vendors/{vendor_id}", status_code=204) async def delete_vendor( vendor_id: str, - db: Session = Depends(get_db), -): + service: BannerAdminService = Depends(get_admin_service), +) -> None: """Delete a vendor.""" - try: - vid = uuid.UUID(vendor_id) - except ValueError: - raise HTTPException(status_code=400, detail="Invalid vendor ID") - - vendor = db.query(BannerVendorConfigDB).filter(BannerVendorConfigDB.id == vid).first() - if not vendor: - raise HTTPException(status_code=404, detail="Vendor not found") - - db.delete(vendor) - db.commit() + with translate_domain_errors(): + service.delete_vendor(vendor_id) diff --git a/backend-compliance/compliance/schemas/banner.py b/backend-compliance/compliance/schemas/banner.py new file mode 100644 index 0000000..27c931a --- /dev/null +++ b/backend-compliance/compliance/schemas/banner.py @@ -0,0 +1,85 @@ +""" +Banner consent schemas — cookie consent SDK + admin configuration. + +Phase 1 Step 4: extracted from ``compliance.api.banner_routes`` so the +route layer becomes thin delegation to ``compliance.services.banner_*``. +""" + +from typing import Any, List, Optional + +from pydantic import BaseModel, ConfigDict + + +class ConsentCreate(BaseModel): + """Request body for recording a device consent.""" + site_id: str + device_fingerprint: str + categories: List[str] = [] + vendors: List[str] = [] + ip_address: Optional[str] = None + user_agent: Optional[str] = None + consent_string: Optional[str] = None + + +class SiteConfigCreate(BaseModel): + """Request body for creating a banner site configuration.""" + site_id: str + site_name: Optional[str] = None + site_url: Optional[str] = None + banner_title: Optional[str] = None + banner_description: Optional[str] = None + privacy_url: Optional[str] = None + imprint_url: Optional[str] = None + dsb_name: Optional[str] = None + dsb_email: Optional[str] = None + theme: Optional[dict[str, Any]] = None + tcf_enabled: bool = False + + +class SiteConfigUpdate(BaseModel): + """Partial update for a banner site configuration.""" + + model_config = ConfigDict(extra="ignore") + + site_name: Optional[str] = None + site_url: Optional[str] = None + banner_title: Optional[str] = None + banner_description: Optional[str] = None + privacy_url: Optional[str] = None + imprint_url: Optional[str] = None + dsb_name: Optional[str] = None + dsb_email: Optional[str] = None + theme: Optional[dict[str, Any]] = None + tcf_enabled: Optional[bool] = None + is_active: Optional[bool] = None + + +class CategoryConfigCreate(BaseModel): + """Request body for adding a cookie category to a site.""" + category_key: str + name_de: str + name_en: Optional[str] = None + description_de: Optional[str] = None + description_en: Optional[str] = None + is_required: bool = False + sort_order: int = 0 + + +class VendorConfigCreate(BaseModel): + """Request body for adding a vendor under a site's category.""" + vendor_name: str + vendor_url: Optional[str] = None + category_key: str + description_de: Optional[str] = None + description_en: Optional[str] = None + cookie_names: List[str] = [] + retention_days: int = 365 + + +__all__ = [ + "ConsentCreate", + "SiteConfigCreate", + "SiteConfigUpdate", + "CategoryConfigCreate", + "VendorConfigCreate", +] diff --git a/backend-compliance/compliance/services/_banner_serializers.py b/backend-compliance/compliance/services/_banner_serializers.py new file mode 100644 index 0000000..3b45825 --- /dev/null +++ b/backend-compliance/compliance/services/_banner_serializers.py @@ -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, + } diff --git a/backend-compliance/compliance/services/banner_admin_service.py b/backend-compliance/compliance/services/banner_admin_service.py new file mode 100644 index 0000000..6a8aa50 --- /dev/null +++ b/backend-compliance/compliance/services/banner_admin_service.py @@ -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() diff --git a/backend-compliance/compliance/services/banner_consent_service.py b/backend-compliance/compliance/services/banner_consent_service.py new file mode 100644 index 0000000..293d92d --- /dev/null +++ b/backend-compliance/compliance/services/banner_consent_service.py @@ -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() + }, + } diff --git a/backend-compliance/mypy.ini b/backend-compliance/mypy.ini index 03bf0ad..abd8118 100644 --- a/backend-compliance/mypy.ini +++ b/backend-compliance/mypy.ini @@ -70,8 +70,10 @@ ignore_errors = True [mypy-compliance.api.*] ignore_errors = True -# Refactored route module under Step 4 — override the blanket rule above. +# Refactored route modules under Step 4 — override the blanket rule above. [mypy-compliance.api.audit_routes] ignore_errors = False +[mypy-compliance.api.banner_routes] +ignore_errors = False [mypy-compliance.api._http_errors] ignore_errors = False diff --git a/backend-compliance/tests/contracts/openapi.baseline.json b/backend-compliance/tests/contracts/openapi.baseline.json index 50fc625..7b48249 100644 --- a/backend-compliance/tests/contracts/openapi.baseline.json +++ b/backend-compliance/tests/contracts/openapi.baseline.json @@ -2008,6 +2008,7 @@ "type": "object" }, "CategoryConfigCreate": { + "description": "Request body for adding a cookie category to a site.", "properties": { "category_key": { "title": "Category Key", @@ -16041,6 +16042,7 @@ "type": "object" }, "SiteConfigCreate": { + "description": "Request body for creating a banner site configuration.", "properties": { "banner_description": { "anyOf": [ @@ -16159,6 +16161,7 @@ "type": "object" }, "SiteConfigUpdate": { + "description": "Partial update for a banner site configuration.", "properties": { "banner_description": { "anyOf": [ @@ -19274,6 +19277,7 @@ "type": "object" }, "VendorConfigCreate": { + "description": "Request body for adding a vendor under a site's category.", "properties": { "category_key": { "title": "Category Key", @@ -19495,73 +19499,6 @@ "title": "VersionResponse", "type": "object" }, - "compliance__api__banner_routes__ConsentCreate": { - "properties": { - "categories": { - "default": [], - "items": { - "type": "string" - }, - "title": "Categories", - "type": "array" - }, - "consent_string": { - "anyOf": [ - { - "type": "string" - }, - { - "type": "null" - } - ], - "title": "Consent String" - }, - "device_fingerprint": { - "title": "Device Fingerprint", - "type": "string" - }, - "ip_address": { - "anyOf": [ - { - "type": "string" - }, - { - "type": "null" - } - ], - "title": "Ip Address" - }, - "site_id": { - "title": "Site Id", - "type": "string" - }, - "user_agent": { - "anyOf": [ - { - "type": "string" - }, - { - "type": "null" - } - ], - "title": "User Agent" - }, - "vendors": { - "default": [], - "items": { - "type": "string" - }, - "title": "Vendors", - "type": "array" - } - }, - "required": [ - "site_id", - "device_fingerprint" - ], - "title": "ConsentCreate", - "type": "object" - }, "compliance__api__einwilligungen_routes__ConsentCreate": { "properties": { "consent_version": { @@ -20353,6 +20290,74 @@ ], "title": "TemplateCreate", "type": "object" + }, + "compliance__schemas__banner__ConsentCreate": { + "description": "Request body for recording a device consent.", + "properties": { + "categories": { + "default": [], + "items": { + "type": "string" + }, + "title": "Categories", + "type": "array" + }, + "consent_string": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Consent String" + }, + "device_fingerprint": { + "title": "Device Fingerprint", + "type": "string" + }, + "ip_address": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Ip Address" + }, + "site_id": { + "title": "Site Id", + "type": "string" + }, + "user_agent": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "User Agent" + }, + "vendors": { + "default": [], + "items": { + "type": "string" + }, + "title": "Vendors", + "type": "array" + } + }, + "required": [ + "site_id", + "device_fingerprint" + ], + "title": "ConsentCreate", + "type": "object" } } }, @@ -21019,7 +21024,11 @@ "200": { "content": { "application/json": { - "schema": {} + "schema": { + "additionalProperties": true, + "title": "Response Delete Audit Session Api Compliance Audit Sessions Session Id Delete", + "type": "object" + } } }, "description": "Successful Response" @@ -21103,7 +21112,11 @@ "200": { "content": { "application/json": { - "schema": {} + "schema": { + "additionalProperties": true, + "title": "Response Archive Audit Session Api Compliance Audit Sessions Session Id Archive Put", + "type": "object" + } } }, "description": "Successful Response" @@ -21145,7 +21158,11 @@ "200": { "content": { "application/json": { - "schema": {} + "schema": { + "additionalProperties": true, + "title": "Response Complete Audit Session Api Compliance Audit Sessions Session Id Complete Put", + "type": "object" + } } }, "description": "Successful Response" @@ -21250,7 +21267,11 @@ "200": { "content": { "application/json": { - "schema": {} + "schema": { + "additionalProperties": true, + "title": "Response Start Audit Session Api Compliance Audit Sessions Session Id Start Put", + "type": "object" + } } }, "description": "Successful Response" @@ -21336,7 +21357,14 @@ "200": { "content": { "application/json": { - "schema": {} + "schema": { + "items": { + "additionalProperties": true, + "type": "object" + }, + "title": "Response List Site Configs Api Compliance Banner Admin Sites Get", + "type": "array" + } } }, "description": "Successful Response" @@ -21393,7 +21421,11 @@ "200": { "content": { "application/json": { - "schema": {} + "schema": { + "additionalProperties": true, + "title": "Response Create Site Config Api Compliance Banner Admin Sites Post", + "type": "object" + } } }, "description": "Successful Response" @@ -21512,7 +21544,11 @@ "200": { "content": { "application/json": { - "schema": {} + "schema": { + "additionalProperties": true, + "title": "Response Update Site Config Api Compliance Banner Admin Sites Site Id Put", + "type": "object" + } } }, "description": "Successful Response" @@ -21570,7 +21606,14 @@ "200": { "content": { "application/json": { - "schema": {} + "schema": { + "items": { + "additionalProperties": true, + "type": "object" + }, + "title": "Response List Categories Api Compliance Banner Admin Sites Site Id Categories Get", + "type": "array" + } } }, "description": "Successful Response" @@ -21636,7 +21679,11 @@ "200": { "content": { "application/json": { - "schema": {} + "schema": { + "additionalProperties": true, + "title": "Response Create Category Api Compliance Banner Admin Sites Site Id Categories Post", + "type": "object" + } } }, "description": "Successful Response" @@ -21694,7 +21741,14 @@ "200": { "content": { "application/json": { - "schema": {} + "schema": { + "items": { + "additionalProperties": true, + "type": "object" + }, + "title": "Response List Vendors Api Compliance Banner Admin Sites Site Id Vendors Get", + "type": "array" + } } }, "description": "Successful Response" @@ -21760,7 +21814,11 @@ "200": { "content": { "application/json": { - "schema": {} + "schema": { + "additionalProperties": true, + "title": "Response Create Vendor Api Compliance Banner Admin Sites Site Id Vendors Post", + "type": "object" + } } }, "description": "Successful Response" @@ -21818,7 +21876,11 @@ "200": { "content": { "application/json": { - "schema": {} + "schema": { + "additionalProperties": true, + "title": "Response Get Site Stats Api Compliance Banner Admin Stats Site Id Get", + "type": "object" + } } }, "description": "Successful Response" @@ -21913,7 +21975,11 @@ "200": { "content": { "application/json": { - "schema": {} + "schema": { + "additionalProperties": true, + "title": "Response Get Site Config Api Compliance Banner Config Site Id Get", + "type": "object" + } } }, "description": "Successful Response" @@ -21980,7 +22046,11 @@ "200": { "content": { "application/json": { - "schema": {} + "schema": { + "additionalProperties": true, + "title": "Response Get Consent Api Compliance Banner Consent Get", + "type": "object" + } } }, "description": "Successful Response" @@ -22027,7 +22097,7 @@ "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/compliance__api__banner_routes__ConsentCreate" + "$ref": "#/components/schemas/compliance__schemas__banner__ConsentCreate" } } }, @@ -22037,7 +22107,11 @@ "200": { "content": { "application/json": { - "schema": {} + "schema": { + "additionalProperties": true, + "title": "Response Record Consent Api Compliance Banner Consent Post", + "type": "object" + } } }, "description": "Successful Response" @@ -22104,7 +22178,11 @@ "200": { "content": { "application/json": { - "schema": {} + "schema": { + "additionalProperties": true, + "title": "Response Export Consent Api Compliance Banner Consent Export Get", + "type": "object" + } } }, "description": "Successful Response" @@ -22162,7 +22240,11 @@ "200": { "content": { "application/json": { - "schema": {} + "schema": { + "additionalProperties": true, + "title": "Response Withdraw Consent Api Compliance Banner Consent Consent Id Delete", + "type": "object" + } } }, "description": "Successful Response"