New version

Author: timaakulich

fastapi_toolkit/application.py

@@ -0,0 +1,24 @@
+from fastapi import (
+    FastAPI,
+    status,
+)
+from fastapi.encoders import jsonable_encoder
+from fastapi.exceptions import RequestValidationError
+from fastapi.responses import JSONResponse
+
+from fastapi_toolkit.exceptions import Error, exc_detail
+
+
+def prepare_app(app: FastAPI) -> FastAPI:
+
+    @app.exception_handler(RequestValidationError)
+    async def validation_exception_handler(request, exc):
+        return JSONResponse({
+            'detail': exc_detail(
+                code=Error.request_validation_error,
+                error='Invalid request',
+                info=jsonable_encoder(exc.errors())
+            )
+        }, status_code=status.HTTP_422_UNPROCESSABLE_ENTITY)
+
+    return app

fastapi_toolkit/dependencies/__init__.py

@@ -0,0 +1,76 @@
+from collections.abc import Awaitable, Callable
+from inspect import iscoroutinefunction
+from typing import TypeVar
+
+import jwt
+from fastapi import (
+    Depends,
+    HTTPException,
+    status,
+)
+from fastapi.security import (
+    HTTPAuthorizationCredentials,
+    SecurityScopes,
+)
+from jwt import PyJWTError
+
+from fastapi_toolkit.exceptions import Error, exc_detail
+from fastapi_toolkit.schemas.user import DecodedUserModel
+from fastapi_toolkit.security import HTTPBearer
+
+T = TypeVar('T', bound=DecodedUserModel)
+
+
+def get_user_dependency(
+        jwt_secret_dependency: Callable[..., str | Awaitable[str]],
+        alg_dependency: Callable[..., str | Awaitable[str]],
+        project_dependency: Callable[..., str | Awaitable[str]],
+        user_model: type[T],
+        token_validator: Callable[[T], None | Awaitable[None]] | None = None,
+) -> Callable[..., Awaitable[T]]:
+    async def get_user(
+            security_scopes: SecurityScopes,
+            auth: HTTPAuthorizationCredentials = Depends(HTTPBearer(
+                auto_error=True
+            )),
+            jwt_secret: str = Depends(jwt_secret_dependency),
+            alg: str = Depends(alg_dependency),
+            project: str = Depends(project_dependency),
+    ) -> T:
+        try:
+            decoded_token = jwt.decode(auth.credentials, jwt_secret, algorithms=[alg])
+        except PyJWTError as exc:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail=exc_detail(
+                    code=Error.jwt_validation_error,
+                    error=str(exc)
+                )
+            ) from exc
+        else:
+            decoded_token = user_model.model_validate(decoded_token)
+
+            if token_validator is not None:
+                if iscoroutinefunction(token_validator):
+                    await token_validator(decoded_token)
+                else:
+                    token_validator(decoded_token)
+
+            user_permissions = decoded_token.permissions.get(project, 0)
+            for scope in (security_scopes.scopes or []):
+                scope = int(scope)
+                if not (user_permissions & scope == scope):
+                    raise HTTPException(
+                        status_code=status.HTTP_403_FORBIDDEN,
+                        detail=exc_detail(
+                            code=Error.permissions_error,
+                            error='Permissions required',
+                            info={
+                                'project': project,
+                                'required_permission': scope,
+                                'user_permissions': user_permissions
+                            }
+                        )
+                    )
+            return decoded_token
+    return get_user

fastapi_toolkit/dependencies/user.py

@@ -0,0 +1,19 @@
+import enum
+from typing import Any
+
+
+class Error(enum.StrEnum):
+    auth_error = 'auth_error'
+    request_validation_error = 'request_validation_error'
+    jwt_validation_error = 'jwt_validation_error'
+    permissions_error = 'permissions_error'
+    object_not_found = 'object_not_found'
+    integrity_error = 'integrity_error'
+
+
+def exc_detail(code: enum.StrEnum, error: str | None = None, info: Any | None = None) -> dict[str, Any]:
+    return {
+        'code': code.value,
+        'error': error,
+        'info': info
+    }

fastapi_toolkit/exceptions.py

@@ -0,0 +1,13 @@
+from typing import Any
+
+from pydantic import BaseModel
+
+
+class DetailModel(BaseModel):
+    code: str
+    error: str | None = None
+    info: Any | None = None
+
+
+class ErrorResponseModel(BaseModel):
+    detail: DetailModel

fastapi_toolkit/schemas/__init__.py

@@ -0,0 +1,13 @@
+from uuid import UUID
+
+from pydantic import (
+    BaseModel,
+    Field,
+)
+
+
+class DecodedUserModel(BaseModel):
+    id: UUID = Field(..., alias="sub")
+    permissions: dict[str, int] = Field(default_factory=dict)
+    version: int = 0
+    jid: UUID | None = None

fastapi_toolkit/schemas/response.py

@@ -0,0 +1,66 @@
+# Standard Library
+import enum
+from typing import Literal
+
+# Third Party Library
+from fastapi import (
+    HTTPException,
+    Request,
+    status,
+)
+from fastapi.security.http import (
+    HTTPAuthorizationCredentials,
+)
+from fastapi.security.http import (
+    HTTPBearer as BaseHTTPBearer,
+)
+from fastapi.security.utils import get_authorization_scheme_param
+
+from fastapi_toolkit.exceptions import Error, exc_detail
+from fastapi_toolkit.schemas.response import DetailModel
+
+
+def get_scopes(*scopes: enum.IntEnum | enum.Flag) -> list[str]:
+    return [str(scope.value) for scope in scopes]
+
+
+class HTTPBearer(BaseHTTPBearer):
+    async def __call__(
+        self, request: Request
+    ) -> HTTPAuthorizationCredentials | None:
+        authorization = request.headers.get('Authorization')
+        scheme, credentials = get_authorization_scheme_param(authorization)
+        if not (authorization and scheme and credentials):
+            if self.auto_error:
+                raise HTTPException(
+                    status_code=status.HTTP_403_FORBIDDEN,
+                    detail=exc_detail(
+                        code=Error.auth_error,
+                        error='Not authenticated'
+                    )
+                )
+            else:
+                return None
+        if scheme.lower() != 'bearer':
+            if self.auto_error:
+                raise HTTPException(
+                    status_code=status.HTTP_403_FORBIDDEN,
+                    detail=exc_detail(
+                        code=Error.auth_error,
+                        error='Invalid authentication credentials'
+                    ),
+                )
+            else:
+                return None
+        return HTTPAuthorizationCredentials(
+            scheme=scheme,
+            credentials=credentials
+        )
+
+
+ERROR_AUTH_RESPONSES = {
+    status.HTTP_403_FORBIDDEN: {
+        'model': DetailModel,
+        'description': 'Not authenticated / Invalid authentication credentials',
+    }
+}

fastapi_toolkit/schemas/user.py

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "fastapi-toolkit"
-version = "0.1.0"
+version = "0.1.1"
 description = "Common toolkit for FastAPI + SQLAlchemy projects: async DB sessions, base CRUD, settings with AWS Secrets Manager"
 readme = "README.md"
 license = "MIT"