Instruction file imported from mick-gsk/PWBS (.github/instructions/backend.instructions.md). Copyright stays with the author.

Backend-Instruktionen: Python / FastAPI

Modul-Struktur

pwbs/
├── api/            # FastAPI Router, Dependency Injection, Middleware
├── connectors/     # Datenquellen-Konnektoren (BaseConnector + Implementierungen)
├── ingestion/      # Ingestion-Pipeline, Normalisierung ins UDF
├── processing/     # Chunking, Embedding, NER
├── storage/        # Repository-Schicht: PostgreSQL, Weaviate, Neo4j
├── briefing/       # Briefing-Generierung, Templates
├── search/         # Semantische Suche, Hybrid-Search
├── graph/          # Knowledge-Graph-Operationen
├── scheduler/      # Zeitgesteuerte Jobs
├── prompts/        # LLM-Prompt-Templates (versioniert)
└── core/           # Shared: Config, Exceptions, Basisklassen

MVP-Fokussierung (ADR-016)

Deaktivierte Module

Folgende Module liegen in backend/_deferred/ und sind nicht Teil des MVP. Code in diesen Modulen NICHT importieren, referenzieren oder weiterentwickeln:

billing – Zahlungen/Subscriptions (Phase 4)
teams – Multi-User-Organisationen (Phase 4)
rbac – Rollenbasierte Zugriffskontrolle (Phase 4)
marketplace – Plugin-Marketplace (Phase 5)
developer – Öffentliche API / API-Keys (Phase 5)
sso – Enterprise SSO/SAML (Phase 4)

Aktive Konnektoren (nur Kern-4)

Im MVP nur Google Calendar, Notion, Zoom, Obsidian aktiv. Phase-3-Konnektoren (Gmail, Slack, Google Docs, Outlook) liegen in backend/_deferred/connectors/ – NICHT importieren oder weiterentwickeln.

Neo4j ist optional

get_neo4j_driver() gibt None zurück wenn Neo4j nicht erreichbar ist.
Jeder Code der Neo4j nutzt, MUSS mit driver is None umgehen können.
NullGraphService-Fallbacks verwenden; keine harte Abhängigkeit.

Pflichtregeln

Typing

Vollständige Type Annotations in jeder Funktion und Methode. Any nur wenn unvermeidbar und dann mit Kommentar begründen.
TypeVar, Generic, Protocol für polymorphe Strukturen nutzen.
list[X] statt List[X], dict[K,V] statt Dict[K,V] (Python 3.12+).

Pydantic v2

Alle Datenmodelle erben von pydantic.BaseModel.
model_config = ConfigDict(...) statt class Config.
@model_validator(mode="after") statt @validator.
@field_validator mit mode="before" für Input-Transformation.
Niemals dict() auf Pydantic-Objekte aufrufen – model_dump() verwenden.

FastAPI

Response-Typ korrekt in Route-Signatur annotieren: response: Response NICHT response: Response | None.
Response-Parameter VOR Default-Parameter-Dependencies platzieren (sonst FastAPIError).
Dependency Injection über Depends() für DB-Sessions und Service-Instanzen.
APIRouter mit prefix und tags in jedem Modul.

Fehlerbehandlung

# Eigene Exception-Hierarchie
class PWBSError(Exception): ...
class ConnectorError(PWBSError): ...
class StorageError(PWBSError): ...

# HTTP-Fehler mit strukturiertem detail
raise HTTPException(
    status_code=422,
    detail={"code": "VALIDATION_FAILED", "field": "source_id", "message": "..."}
)

Async & Threading

async def für alle I/O-Operationen (DB-Zugriff, HTTP-Calls, File-I/O).
Synchroner Blocking-Code (z.B. schwere Berechnungen, sync-Bibliotheken) in asyncio.to_thread() auslagern.
DB-Sessions niemals über Modul-Grenzen teilen.

Idempotenz (KRITISCH)

Jeder ingest()- und process()-Aufruf muss bei Wiederholung dasselbe Ergebnis liefern ohne Duplikate.
ON CONFLICT DO UPDATE (upsert) in PostgreSQL-Writes nutzen.
Vor jedem Weaviate-Insert prüfen, ob source_id + owner_id bereits existiert.
Cursor/Watermarks in Connector-Zuständen persistieren.

DSGVO-Pflichten

Jedes UnifiedDocument braucht owner_id und expires_at.
Jede DB-Query gegen UnifiedDocument oder abgeleitete Tabellen MUSS WHERE owner_id = :user_id enthalten.
Keine Nutzerdaten in Logs (kein content, keine Embeddings, keine metadata-Werte).

Tests

Fixtures für alle externen Abhängigkeiten in conftest.py.
Kein echter Netzwerkzugriff im Unit-Test – httpx.MockTransport oder pytest-mock.
@pytest.mark.asyncio für async Tests, asyncio_mode = "auto" in pytest.ini.
Test-Dateipfad: tests/unit/, tests/integration/, tests/e2e/.

Konnektoren implementieren

from pwbs.connectors.base import BaseConnector, ConnectorConfig, SyncResult

class MyConnector(BaseConnector):
    async def fetch_since(self, cursor: str | None) -> SyncResult:
        """Cursor-basiertes Abrufen. Gibt neuen Cursor zurück."""
        ...

    async def normalize(self, raw: dict) -> UnifiedDocument:
        """Rohdaten → UDF. Muss idempotent sein."""
        ...

Reasoning-Anforderungen (Claude Opus 4.6)

Bei jeder Implementierungsaufgabe den Denkprozess explizit durchlaufen:

Vor dem Code

Invarianten identifizieren: Welche Bedingungen müssen vor und nach dem Methodenaufruf gelten?
Fehlerszenarien durchdenken: Was passiert bei leerem Input, abgelaufenem Token, Netzwerkausfall, DB-Timeout?
Abhängigkeiten kartieren: Welche anderen Module werden durch die Änderung beeinflusst?
Breaking Changes beurteilen: Ändert sich ein bestehender Contract (Signatur, Verhalten, Rückgabetyp)?

Selbst-Review nach dem Code

Vor dem Ausgeben generierten Code intern prüfen:

Alle owner_id-Filter vorhanden? Gibt es Queries ohne Mandanten-Isolation?
Alle Writes als Upsert/idempotent implementiert?
Alle I/O-Operationen als async def?
Kein PII in Logging-Aufrufen?
Rückgabetypen vollständig und korrekt annotiert?

Vollständigkeitsgebot

Keine pass-Stubs oder # TODO: implement – entweder vollständig implementieren oder explizit raise NotImplementedError("Begründung").
Alle Randfälle explizit behandeln: leere Listen, None-Werte, leere Strings, Maximalwert-Überschreitungen.


## LLM-Aufrufe

- Prompts ausschließlich aus `pwbs/prompts/*.jinja2` laden, nie inline hardcoden.
- Structured Output (JSON-Schema) für alle nicht-narrativen LLM-Outputs.
- Jede LLM-Antwort mit Quellenreferenzen versehen (Pflichtfeld `sources: list[SourceRef]`).
- LLM-Calls über den `LLMGateway`-Service abstrahieren – nie direkt Anthropic/OpenAI SDK aufrufen.
- Fallback-Reihenfolge: Claude → GPT-4 → Ollama (konfigurierbar per Env-Var).

---

## REST API Konventionen

### Endpunkt-Design

| Operation | HTTP-Methode | Pfad-Muster | Status-Code |
|-----------|--------------|-------------|-------------|
| Liste abrufen | `GET` | `/resources` | 200 |
| Einzelne Ressource | `GET` | `/resources/{id}` | 200 / 404 |
| Erstellen | `POST` | `/resources` | 201 (Location-Header!) |
| Aktualisieren | `PATCH` | `/resources/{id}` | 200 / 404 |
| Vollständig ersetzen | `PUT` | `/resources/{id}` | 200 / 404 |
| Löschen | `DELETE` | `/resources/{id}` | 204 (kein Body!) / 404 |

### Namenskonventionen

- **Plural** für Sammlungen: `/briefings`, nicht `/briefing`
- **kebab-case** für mehrteilige Namen: `/search-history`, nicht `/searchHistory`
- **Ressourcen-IDs**: UUIDv4, nie sequentielle Integer
- **Verschachtelung max. 2 Ebenen**: `/connectors/{id}/sync-runs`, nicht `/users/{uid}/connectors/{cid}/sync-runs/{sid}/logs`

### Pagination (bei Listen)

```python
@router.get("/resources")
async def list_resources(
    skip: int = Query(0, ge=0),
    limit: int = Query(50, ge=1, le=100),
    sort_by: str = Query("created_at"),
    sort_order: Literal["asc", "desc"] = Query("desc"),
) -> PaginatedResponse[ResourceResponse]:
    ...

Response enthält items, total, skip, limit.

Fehlerformat (RFC 7807-inspiriert)

{
  "code": "VALIDATION_ERROR",
  "message": "Human-readable description",
  "field": "source_id",  // bei Validierungsfehlern
  "details": { ... }     // optional, für Debug-Kontext
}

Kein Stack-Trace in Produktionsfehlerantworten!

Versionierung

URL-Präfix: /api/v1/...
Keine Breaking Changes ohne Major-Version-Bump
Deprecation-Header: Deprecation: true, Sunset: <date>

OpenAPI / Swagger

Jede Route braucht summary und description
response_model explizit annotieren – kein Any
Tags (tags=["Briefings"]) für API-Explorer-Gruppierung

backend

About