refactor(api): introduce UserContext schema for API dependency injection

Replace direct User database model imports in API endpoints with UserContext
schema, following the architecture principle that API routes should not import
database models directly.

Changes:
- Create UserContext schema in models/schema/auth.py with from_user() factory
- Update app/api/deps.py to return UserContext from all auth dependencies
- Add _get_user_model() helper for functions needing User model access
- Update 58 API endpoint files to use UserContext instead of User
- Add noqa comments for 4 legitimate edge cases (enums, internal helpers)

Architecture validation: 0 errors (down from 61), 11 warnings remain

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2026-01-30 20:47:33 +01:00
parent 1ad30bd77e
commit cad862f469
60 changed files with 755 additions and 589 deletions

View File

@@ -19,7 +19,7 @@ from app.services.admin_notification_service import (
admin_notification_service,
platform_alert_service,
)
from models.database.user import User
from models.schema.auth import UserContext
from models.schema.admin import (
AdminNotificationCreate,
AdminNotificationListResponse,
@@ -52,7 +52,7 @@ def get_notifications(
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=100),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
current_admin: UserContext = Depends(get_current_admin_api),
) -> AdminNotificationListResponse:
"""Get admin notifications with filtering."""
notifications, total, unread_count = admin_notification_service.get_notifications(
@@ -93,7 +93,7 @@ def get_notifications(
def create_notification(
notification_data: AdminNotificationCreate,
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
current_admin: UserContext = Depends(get_current_admin_api),
) -> AdminNotificationResponse:
"""Create a new admin notification (manual)."""
notification = admin_notification_service.create_from_schema(
@@ -123,7 +123,7 @@ def create_notification(
def get_recent_notifications(
limit: int = Query(5, ge=1, le=10),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
current_admin: UserContext = Depends(get_current_admin_api),
) -> dict:
"""Get recent unread notifications for header dropdown."""
notifications = admin_notification_service.get_recent_notifications(
@@ -151,7 +151,7 @@ def get_recent_notifications(
@router.get("/unread-count", response_model=UnreadCountResponse)
def get_unread_count(
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
current_admin: UserContext = Depends(get_current_admin_api),
) -> UnreadCountResponse:
"""Get count of unread notifications."""
count = admin_notification_service.get_unread_count(db)
@@ -162,7 +162,7 @@ def get_unread_count(
def mark_as_read(
notification_id: int,
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
current_admin: UserContext = Depends(get_current_admin_api),
) -> MessageResponse:
"""Mark notification as read."""
notification = admin_notification_service.mark_as_read(
@@ -178,7 +178,7 @@ def mark_as_read(
@router.put("/mark-all-read", response_model=MessageResponse)
def mark_all_as_read(
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
current_admin: UserContext = Depends(get_current_admin_api),
) -> MessageResponse:
"""Mark all notifications as read."""
count = admin_notification_service.mark_all_as_read(
@@ -193,7 +193,7 @@ def mark_all_as_read(
def delete_notification(
notification_id: int,
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
current_admin: UserContext = Depends(get_current_admin_api),
) -> MessageResponse:
"""Delete a notification."""
deleted = admin_notification_service.delete_notification(
@@ -220,7 +220,7 @@ def get_platform_alerts(
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=100),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
current_admin: UserContext = Depends(get_current_admin_api),
) -> PlatformAlertListResponse:
"""Get platform alerts with filtering."""
alerts, total, active_count, critical_count = platform_alert_service.get_alerts(
@@ -266,7 +266,7 @@ def get_platform_alerts(
def create_platform_alert(
alert_data: PlatformAlertCreate,
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
current_admin: UserContext = Depends(get_current_admin_api),
) -> PlatformAlertResponse:
"""Create new platform alert (manual)."""
alert = platform_alert_service.create_from_schema(db=db, data=alert_data)
@@ -299,7 +299,7 @@ def resolve_platform_alert(
alert_id: int,
resolve_data: PlatformAlertResolve,
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
current_admin: UserContext = Depends(get_current_admin_api),
) -> MessageResponse:
"""Resolve platform alert."""
alert = platform_alert_service.resolve_alert(
@@ -320,7 +320,7 @@ def resolve_platform_alert(
@router.get("/alerts/stats", response_model=AlertStatisticsResponse)
def get_alert_statistics(
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
current_admin: UserContext = Depends(get_current_admin_api),
) -> AlertStatisticsResponse:
"""Get alert statistics for dashboard."""
stats = platform_alert_service.get_statistics(db)