File size: 2,215 Bytes
b6b458c 97c535b b6b458c e24398c b6b458c 97c535b b6b458c 97c535b b6b458c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 |
"""
API dependencies for dependency injection.
Provides common dependencies used across API routes.
"""
from typing import Optional, Dict, Any, AsyncGenerator
from fastapi import Request, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from src.api.middleware.authentication import get_current_user as _get_current_user
from src.db.session import get_session as get_db_session
def get_current_user(request: Request) -> Dict[str, Any]:
"""
Get current authenticated user from request.
Returns user information stored in request state.
"""
return _get_current_user(request)
def get_current_optional_user(request: Request) -> Optional[Dict[str, Any]]:
"""
Get current user if authenticated, None otherwise.
Used for endpoints that support both authenticated and anonymous access.
"""
try:
# Check if user is authenticated
if hasattr(request.state, "user_id") and request.state.user_id:
return get_current_user(request)
return None
except Exception:
# If any error occurs, treat as anonymous user
return None
async def get_db() -> AsyncGenerator[AsyncSession, None]:
"""
Get database session.
Yields an async SQLAlchemy session.
"""
async with get_db_session() as session:
yield session
def require_admin(user: Dict[str, Any] = Depends(get_current_user)) -> Dict[str, Any]:
"""
Require admin role for access.
Raises 403 if user is not admin.
"""
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Not authenticated"
)
# Check for admin role
user_role = user.get("role", "").lower()
is_admin = user.get("is_admin", False)
is_superuser = user.get("is_superuser", False)
if user_role != "admin" and not is_admin and not is_superuser:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Admin privileges required"
)
return user
# Export commonly used dependencies
__all__ = [
"get_current_user",
"get_current_optional_user",
"get_db",
"require_admin",
] |