refactor: migrate to Annotated type aliases for FastAPI dependencies

README.md

@@ -33,6 +33,7 @@
 * ⚡️ Fully async FastAPI + SQLAlchemy 2.0
 * 🧱 Pydantic v2 models & validation
 * 🔐 Server-side sessions + CSRF; OAuth (Google wired, GitHub scaffolded); API keys
+* 🏷️ Annotated type aliases for all FastAPI dependencies
 * 👮 Rate limiter with per-tier, per-path rules
 * 🧰 FastCRUD for efficient CRUD & pagination
 * 🧑‍💼 **SQLAdmin**-based admin panel (optional, env-toggled)

backend/src/infrastructure/auth/routes.py

@@ -1,25 +1,27 @@
 import inspect
-from typing import Annotated, Any
+from typing import Any
 
-from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response, status
+from fastapi import APIRouter, HTTPException, Query, Request, Response, status
 from fastapi.responses import RedirectResponse
-from fastapi.security import OAuth2PasswordRequestForm
-from sqlalchemy.ext.asyncio import AsyncSession
 
 from ...modules.user.crud import crud_users
 from ...modules.user.enums import OAuthProvider
 from ..config.settings import get_settings
-from ..database.session import async_session
+from ..dependencies import (
+    AsyncSessionDep,
+    CurrentSessionDataDep,
+    GoogleOAuthProviderDep,
+    OAuth2FormDep,
+    OAuthStateStorageDep,
+    OptionalSessionDataDep,
+    SessionManagerDep,
+)
 from ..logging import get_logger
 from .http_exceptions import UnauthorizedException
-from .oauth.dependencies import get_google_provider, get_oauth_state, get_oauth_state_storage
-from .oauth.provider import AbstractOAuthProvider
+from .oauth.dependencies import get_oauth_state
 from .oauth.schemas import OAuthState, OAuthToken
 from .oauth.services import oauth_account_service
-from .session.dependencies import authenticate_user, get_current_session_data, get_session_manager
-from .session.manager import SessionManager
-from .session.schemas import SessionData
-from .session.storage import AbstractSessionStorage
+from .session.dependencies import authenticate_user
 
 settings = get_settings()
 logger = get_logger()
@@ -52,9 +54,9 @@
 async def login(
     request: Request,
     response: Response,
-    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
-    db: Annotated[AsyncSession, Depends(async_session)],
-    session_manager: Annotated[SessionManager, Depends(get_session_manager)],
+    form_data: OAuth2FormDep,
+    db: AsyncSessionDep,
+    session_manager: SessionManagerDep,
 ) -> dict[str, str]:
     """Login endpoint to get session cookies.
 
@@ -124,8 +126,8 @@ async def login(
 async def logout(
     request: Request,
     response: Response,
-    session_data: Annotated[SessionData, Depends(get_current_session_data)],
-    session_manager: Annotated[SessionManager, Depends(get_session_manager)],
+    session_data: CurrentSessionDataDep,
+    session_manager: SessionManagerDep,
 ) -> dict[str, str]:
     """Logout endpoint to terminate the session and clear cookies."""
     await session_manager.terminate_session(session_data.session_id)
@@ -153,8 +155,8 @@ async def logout(
 async def refresh_csrf_token(
     request: Request,
     response: Response,
-    session_data: Annotated[SessionData, Depends(get_current_session_data)],
-    session_manager: Annotated[SessionManager, Depends(get_session_manager)],
+    session_data: CurrentSessionDataDep,
+    session_manager: SessionManagerDep,
 ) -> dict[str, str]:
     """Generate a new CSRF token for the current session."""
     csrf_token = await session_manager.regenerate_csrf_token(
@@ -201,9 +203,9 @@ async def refresh_csrf_token(
 )
 async def oauth_google_login(
     request: Request,
+    oauth_provider: GoogleOAuthProviderDep,
+    state_storage: OAuthStateStorageDep,
     redirect_uri: str | None = Query(None),
-    oauth_provider: AbstractOAuthProvider = Depends(get_google_provider),
-    state_storage: AbstractSessionStorage[OAuthState] = Depends(get_oauth_state_storage),
 ) -> dict[str, str]:
     """
     Initiate OAuth login flow for Google.
@@ -298,13 +300,13 @@ def _is_provider_valid(provider_value: Any, expected_provider: str) -> bool:
 async def oauth_google_callback(
     request: Request,
     response: Response,
+    oauth_provider: GoogleOAuthProviderDep,
+    state_storage: OAuthStateStorageDep,
+    db: AsyncSessionDep,
+    session_manager: SessionManagerDep,
     code: str = Query(...),
     state: str = Query(...),
     response_format: str = Query("redirect", description="Response format, either 'redirect' or 'json'"),
-    oauth_provider: AbstractOAuthProvider = Depends(get_google_provider),
-    state_storage: AbstractSessionStorage[OAuthState] = Depends(get_oauth_state_storage),
-    db: AsyncSession = Depends(async_session),
-    session_manager: SessionManager = Depends(get_session_manager),
 ):
     """
     Handle OAuth callback from Google.
@@ -421,8 +423,8 @@ async def oauth_google_callback(
 
 @router.get("/check-auth")
 async def check_auth(
-    session_data: Annotated[SessionData | None, Depends(get_current_session_data)],
-    db: AsyncSession = Depends(async_session),
+    session_data: OptionalSessionDataDep,
+    db: AsyncSessionDep,
 ) -> dict[str, Any]:
     """
     Check if the user is authenticated and return basic user information.

backend/src/infrastructure/dependencies.py

@@ -0,0 +1,39 @@
+from typing import Annotated, Any
+
+from fastapi import Depends
+from fastapi.security import OAuth2PasswordRequestForm
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from .auth.oauth.dependencies import get_google_provider, get_oauth_state_storage
+from .auth.oauth.provider import AbstractOAuthProvider
+from .auth.oauth.schemas import OAuthState
+from .auth.session.dependencies import (
+    get_current_session_data,
+    get_current_superuser,
+    get_current_user,
+    get_optional_user,
+    get_session_from_cookie,
+    get_session_manager,
+)
+from .auth.session.manager import SessionManager
+from .auth.session.schemas import SessionData
+from .auth.session.storage import AbstractSessionStorage
+from .database.session import async_session
+
+# Database
+AsyncSessionDep = Annotated[AsyncSession, Depends(async_session)]
+
+# Users
+CurrentUserDep = Annotated[dict[str, Any], Depends(get_current_user)]
+CurrentSuperUserDep = Annotated[dict[str, Any], Depends(get_current_superuser)]
+OptionalUserDep = Annotated[dict[str, Any] | None, Depends(get_optional_user)]
+
+# Sessions
+SessionManagerDep = Annotated[SessionManager, Depends(get_session_manager)]
+CurrentSessionDataDep = Annotated[SessionData, Depends(get_current_session_data)]
+OptionalSessionDataDep = Annotated[SessionData | None, Depends(get_session_from_cookie)]
+
+# OAuth
+OAuth2FormDep = Annotated[OAuth2PasswordRequestForm, Depends()]
+GoogleOAuthProviderDep = Annotated[AbstractOAuthProvider, Depends(get_google_provider)]
+OAuthStateStorageDep = Annotated[AbstractSessionStorage[OAuthState], Depends(get_oauth_state_storage)]

backend/src/modules/api_keys/dependencies.py

@@ -0,0 +1,12 @@
+from typing import Annotated
+
+from fastapi import Depends
+
+from .service import APIKeyService
+
+
+def get_api_key_service() -> APIKeyService:
+    return APIKeyService()
+
+
+APIKeyServiceDep = Annotated[APIKeyService, Depends(get_api_key_service)]

backend/src/modules/api_keys/routes.py

@@ -2,34 +2,26 @@
 
 from typing import Any
 
-from fastapi import APIRouter, Depends, HTTPException, Path, Query, status
+from fastapi import APIRouter, HTTPException, Path, Query, status
 from fastcrud import PaginatedListResponse, compute_offset, paginated_response
-from sqlalchemy.ext.asyncio import AsyncSession
 
-from ...infrastructure.auth.session.dependencies import get_current_user
-from ...infrastructure.database.session import async_session
+from ...infrastructure.dependencies import AsyncSessionDep, CurrentUserDep
 from ..common.exceptions import (
     PermissionDeniedError,
     ResourceNotFoundError,
 )
 from ..common.utils.error_handler import handle_exception
-from ..user.models import User
+from .dependencies import APIKeyServiceDep
 from .schemas import (
     APIKeyCreate,
     APIKeyRead,
     APIKeyUpdate,
     KeyUsageRead,
 )
-from .service import APIKeyService
 
 router = APIRouter(tags=["API Keys"])
 
 
-def get_api_key_service() -> APIKeyService:
-    """Dependency for providing an APIKeyService instance."""
-    return APIKeyService()
-
-
 @router.post(
     "/",
     status_code=201,
@@ -57,9 +49,9 @@ def get_api_key_service() -> APIKeyService:
 )
 async def create_api_key(
     key_data: APIKeyCreate,
-    current_user: User = Depends(get_current_user),
-    api_key_service: APIKeyService = Depends(get_api_key_service),
-    db: AsyncSession = Depends(async_session),
+    current_user: CurrentUserDep,
+    api_key_service: APIKeyServiceDep,
+    db: AsyncSessionDep,
 ) -> dict[str, Any]:
     """Create a new API key for the authenticated user."""
     try:
@@ -96,9 +88,9 @@ async def create_api_key(
     response_description="Paginated list of user's API keys",
 )
 async def get_user_api_keys(
-    current_user: User = Depends(get_current_user),
-    api_key_service: APIKeyService = Depends(get_api_key_service),
-    db: AsyncSession = Depends(async_session),
+    current_user: CurrentUserDep,
+    api_key_service: APIKeyServiceDep,
+    db: AsyncSessionDep,
     active_only: bool = Query(True, description="Return only active keys"),
     page: int = Query(1, ge=1, description="Page number"),
     items_per_page: int = Query(50, ge=1, le=100, description="Items per page"),
@@ -145,10 +137,10 @@ async def get_user_api_keys(
     response_description="API key details",
 )
 async def get_api_key(
+    current_user: CurrentUserDep,
+    api_key_service: APIKeyServiceDep,
+    db: AsyncSessionDep,
     key_id: int = Path(..., description="API key ID"),
-    current_user: User = Depends(get_current_user),
-    api_key_service: APIKeyService = Depends(get_api_key_service),
-    db: AsyncSession = Depends(async_session),
 ) -> dict[str, Any]:
     """Get details for a specific API key."""
     try:
@@ -190,10 +182,10 @@ async def get_api_key(
 )
 async def update_api_key(
     update_data: APIKeyUpdate,
+    current_user: CurrentUserDep,
+    api_key_service: APIKeyServiceDep,
+    db: AsyncSessionDep,
     key_id: int = Path(..., description="API key ID"),
-    current_user: User = Depends(get_current_user),
-    api_key_service: APIKeyService = Depends(get_api_key_service),
-    db: AsyncSession = Depends(async_session),
 ) -> dict[str, Any]:
     """Update an existing API key."""
     try:
@@ -236,10 +228,10 @@ async def update_api_key(
     },
 )
 async def delete_api_key(
+    current_user: CurrentUserDep,
+    api_key_service: APIKeyServiceDep,
+    db: AsyncSessionDep,
     key_id: int = Path(..., description="API key ID"),
-    current_user: User = Depends(get_current_user),
-    api_key_service: APIKeyService = Depends(get_api_key_service),
-    db: AsyncSession = Depends(async_session),
 ) -> None:
     """Delete (deactivate) an API key."""
     try:
@@ -282,10 +274,10 @@ async def delete_api_key(
     response_description="Paginated list of usage records",
 )
 async def get_key_usage(
+    current_user: CurrentUserDep,
+    api_key_service: APIKeyServiceDep,
+    db: AsyncSessionDep,
     key_id: int = Path(..., description="API key ID"),
-    current_user: User = Depends(get_current_user),
-    api_key_service: APIKeyService = Depends(get_api_key_service),
-    db: AsyncSession = Depends(async_session),
     page: int = Query(1, ge=1, description="Page number"),
     items_per_page: int = Query(100, ge=1, le=1000, description="Items per page"),
 ) -> dict[str, Any]:
@@ -340,10 +332,10 @@ async def get_key_usage(
     response_description="Usage analytics for the API key",
 )
 async def get_key_analytics(
+    current_user: CurrentUserDep,
+    api_key_service: APIKeyServiceDep,
+    db: AsyncSessionDep,
     key_id: int = Path(..., description="API key ID"),
-    current_user: User = Depends(get_current_user),
-    api_key_service: APIKeyService = Depends(get_api_key_service),
-    db: AsyncSession = Depends(async_session),
     days: int = Query(30, ge=1, le=365, description="Number of days to analyze"),
 ) -> dict[str, Any]:
     """Get usage analytics for an API key."""
@@ -386,9 +378,9 @@ async def get_key_analytics(
     response_description="Comprehensive API key summary for the user",
 )
 async def get_user_summary(
-    current_user: User = Depends(get_current_user),
-    api_key_service: APIKeyService = Depends(get_api_key_service),
-    db: AsyncSession = Depends(async_session),
+    current_user: CurrentUserDep,
+    api_key_service: APIKeyServiceDep,
+    db: AsyncSessionDep,
 ) -> dict[str, Any]:
     """Get comprehensive API key summary for the authenticated user."""
     try:

backend/src/modules/rate_limit/dependencies.py

@@ -0,0 +1,12 @@
+from typing import Annotated
+
+from fastapi import Depends
+
+from .service import RateLimitService
+
+
+def get_rate_limit_service() -> RateLimitService:
+    return RateLimitService()
+
+
+RateLimitServiceDep = Annotated[RateLimitService, Depends(get_rate_limit_service)]