feat(sdk): API-Referenz Frontend + Backend-Konsolidierung (Shared Utilities, CRUD Factory)
All checks were successful
CI / go-lint (push) Has been skipped
CI / python-lint (push) Has been skipped
CI / nodejs-lint (push) Has been skipped
CI / test-go-ai-compliance (push) Successful in 32s
CI / test-python-backend-compliance (push) Successful in 30s
CI / test-python-document-crawler (push) Successful in 21s
CI / test-python-dsms-gateway (push) Successful in 18s

- API-Referenz Seite (/sdk/api-docs) mit ~690 Endpoints, Suche, Filter, Modul-Index
- Shared db_utils.py (row_to_dict) + tenant_utils Integration in 6 Route-Dateien
- CRUD Factory (crud_factory.py) fuer zukuenftige Module
- Version-Route Auto-Registration in versioning_utils.py
- 1338 Tests bestanden, -232 Zeilen Duplikat-Code

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Benjamin Admin
2026-03-07 17:07:43 +01:00
parent 7ec6b9f6c0
commit 6509e64dd9
19 changed files with 1921 additions and 390 deletions

View File

@@ -13,19 +13,18 @@ import logging
from datetime import datetime
from typing import Optional, Any, Dict
from fastapi import APIRouter, Depends, HTTPException, Query, Header
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel
from sqlalchemy import text
from sqlalchemy.orm import Session
from uuid import UUID
from classroom_engine.database import get_db
from .tenant_utils import get_tenant_id as _get_tenant_id
from .db_utils import row_to_dict as _row_to_dict
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/security-backlog", tags=["security-backlog"])
DEFAULT_TENANT_ID = "9282a473-5c95-4b3a-bf78-0ecc0ec71d3e"
# =============================================================================
# Pydantic Schemas
@@ -61,25 +60,6 @@ class SecurityItemUpdate(BaseModel):
remediation: Optional[str] = None
def _row_to_dict(row) -> Dict[str, Any]:
result = dict(row._mapping)
for key, val in result.items():
if isinstance(val, datetime):
result[key] = val.isoformat()
elif hasattr(val, '__str__') and not isinstance(val, (str, int, float, bool, list, dict, type(None))):
result[key] = str(val)
return result
def _get_tenant_id(x_tenant_id: Optional[str] = Header(None)) -> str:
if x_tenant_id:
try:
UUID(x_tenant_id)
return x_tenant_id
except ValueError:
pass
return DEFAULT_TENANT_ID
# =============================================================================
# Routes
@@ -94,10 +74,9 @@ async def list_security_items(
limit: int = Query(100, ge=1, le=500),
offset: int = Query(0, ge=0),
db: Session = Depends(get_db),
x_tenant_id: Optional[str] = Header(None),
tenant_id: str = Depends(_get_tenant_id),
):
"""List security backlog items with optional filters."""
tenant_id = _get_tenant_id(x_tenant_id)
where_clauses = ["tenant_id = :tenant_id"]
params: Dict[str, Any] = {"tenant_id": tenant_id, "limit": limit, "offset": offset}
@@ -155,10 +134,9 @@ async def list_security_items(
@router.get("/stats")
async def get_security_stats(
db: Session = Depends(get_db),
x_tenant_id: Optional[str] = Header(None),
tenant_id: str = Depends(_get_tenant_id),
):
"""Return security backlog counts."""
tenant_id = _get_tenant_id(x_tenant_id)
rows = db.execute(text("""
SELECT
@@ -189,10 +167,9 @@ async def get_security_stats(
async def create_security_item(
payload: SecurityItemCreate,
db: Session = Depends(get_db),
x_tenant_id: Optional[str] = Header(None),
tenant_id: str = Depends(_get_tenant_id),
):
"""Create a new security backlog item."""
tenant_id = _get_tenant_id(x_tenant_id)
row = db.execute(text("""
INSERT INTO compliance_security_backlog
@@ -226,10 +203,9 @@ async def update_security_item(
item_id: str,
payload: SecurityItemUpdate,
db: Session = Depends(get_db),
x_tenant_id: Optional[str] = Header(None),
tenant_id: str = Depends(_get_tenant_id),
):
"""Update a security backlog item."""
tenant_id = _get_tenant_id(x_tenant_id)
updates: Dict[str, Any] = {"id": item_id, "tenant_id": tenant_id, "updated_at": datetime.utcnow()}
set_clauses = ["updated_at = :updated_at"]
@@ -258,9 +234,8 @@ async def update_security_item(
async def delete_security_item(
item_id: str,
db: Session = Depends(get_db),
x_tenant_id: Optional[str] = Header(None),
tenant_id: str = Depends(_get_tenant_id),
):
tenant_id = _get_tenant_id(x_tenant_id)
result = db.execute(text("""
DELETE FROM compliance_security_backlog
WHERE id = :id AND tenant_id = :tenant_id