275 lines
9.3 KiB
Python
275 lines
9.3 KiB
Python
"""
|
|
OpenAPI Schema Creator Module
|
|
|
|
This module provides functionality to create and customize OpenAPI documentation:
|
|
- Custom security schemes (Bearer Auth, API Key)
|
|
- Response schemas and examples
|
|
- Tag management and descriptions
|
|
- Error responses and validation
|
|
- Custom documentation extensions
|
|
"""
|
|
|
|
from typing import Any, Dict, List, Optional, Set
|
|
from fastapi import FastAPI, APIRouter
|
|
from fastapi.routing import APIRoute
|
|
from fastapi.openapi.utils import get_openapi
|
|
|
|
from AllConfigs.Token.config import Auth
|
|
from AllConfigs.main import MainConfig as Config
|
|
from create_routes import get_all_routers
|
|
|
|
|
|
class OpenAPISchemaCreator:
|
|
"""
|
|
OpenAPI schema creator and customizer for FastAPI applications.
|
|
"""
|
|
|
|
def __init__(self, app: FastAPI):
|
|
"""
|
|
Initialize the OpenAPI schema creator.
|
|
|
|
Args:
|
|
app: FastAPI application instance
|
|
"""
|
|
self.app = app
|
|
_, self.protected_routes = get_all_routers()
|
|
self.tags_metadata = self._create_tags_metadata()
|
|
|
|
@staticmethod
|
|
def _create_tags_metadata() -> List[Dict[str, str]]:
|
|
"""
|
|
Create metadata for API tags.
|
|
|
|
Returns:
|
|
List[Dict[str, str]]: List of tag metadata
|
|
"""
|
|
return [
|
|
{
|
|
"name": "Authentication",
|
|
"description": "Operations related to user authentication and authorization",
|
|
},
|
|
{
|
|
"name": "Users",
|
|
"description": "User management and profile operations",
|
|
},
|
|
# Add more tags as needed
|
|
]
|
|
|
|
def _create_security_schemes(self) -> Dict[str, Any]:
|
|
"""
|
|
Create security scheme definitions.
|
|
|
|
Returns:
|
|
Dict[str, Any]: Security scheme configurations
|
|
"""
|
|
return {
|
|
"Bearer Auth": {
|
|
"type": "apiKey",
|
|
"in": "header",
|
|
"name": Auth.ACCESS_TOKEN_TAG,
|
|
"description": "Enter: **'Bearer <JWT>'**, where JWT is the access token",
|
|
}
|
|
}
|
|
|
|
def _create_common_responses(self) -> Dict[str, Any]:
|
|
"""
|
|
Create common response schemas.
|
|
|
|
Returns:
|
|
Dict[str, Any]: Common response configurations
|
|
"""
|
|
return {
|
|
"401": {
|
|
"description": "Unauthorized - Invalid or missing credentials",
|
|
"content": {
|
|
"application/json": {
|
|
"schema": {"$ref": "#/components/schemas/HTTPValidationError"}
|
|
}
|
|
},
|
|
},
|
|
"403": {
|
|
"description": "Forbidden - Insufficient permissions",
|
|
"content": {
|
|
"application/json": {
|
|
"schema": {"$ref": "#/components/schemas/HTTPValidationError"}
|
|
}
|
|
},
|
|
},
|
|
"422": {
|
|
"description": "Validation Error",
|
|
"content": {
|
|
"application/json": {
|
|
"schema": {"$ref": "#/components/schemas/HTTPValidationError"}
|
|
}
|
|
},
|
|
},
|
|
"500": {
|
|
"description": "Internal Server Error",
|
|
"content": {
|
|
"application/json": {
|
|
"schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"detail": {"type": "string"},
|
|
"error_code": {"type": "string"},
|
|
},
|
|
},
|
|
"example": {
|
|
"detail": "Internal server error occurred",
|
|
"error_code": "INTERNAL_ERROR",
|
|
},
|
|
}
|
|
},
|
|
},
|
|
}
|
|
|
|
def _process_request_body(
|
|
self, path: str, method: str, schema: Dict[str, Any]
|
|
) -> None:
|
|
"""
|
|
Process request body to include examples from model config.
|
|
|
|
Args:
|
|
path: Route path
|
|
method: HTTP method
|
|
schema: OpenAPI schema to modify
|
|
"""
|
|
try:
|
|
route_schema = schema["paths"][path][method]
|
|
if "requestBody" in route_schema:
|
|
request_body = route_schema["requestBody"]
|
|
if "content" in request_body:
|
|
content = request_body["content"]
|
|
if "application/json" in content:
|
|
json_content = content["application/json"]
|
|
if (
|
|
"schema" in json_content
|
|
and "$ref" in json_content["schema"]
|
|
):
|
|
ref = json_content["schema"]["$ref"]
|
|
model_name = ref.split("/")[-1]
|
|
if model_name in schema["components"]["schemas"]:
|
|
model_schema = schema["components"]["schemas"][
|
|
model_name
|
|
]
|
|
if "example" in model_schema:
|
|
json_content["example"] = model_schema["example"]
|
|
except KeyError:
|
|
pass
|
|
|
|
def _process_response_examples(
|
|
self, path: str, method: str, schema: Dict[str, Any]
|
|
) -> None:
|
|
"""
|
|
Process response body to include examples from model config.
|
|
|
|
Args:
|
|
path: Route path
|
|
method: HTTP method
|
|
schema: OpenAPI schema to modify
|
|
"""
|
|
try:
|
|
route_schema = schema["paths"][path][method]
|
|
if "responses" in route_schema:
|
|
responses = route_schema["responses"]
|
|
if "200" in responses:
|
|
response = responses["200"]
|
|
if "content" in response:
|
|
content = response["content"]
|
|
if "application/json" in content:
|
|
json_content = content["application/json"]
|
|
if (
|
|
"schema" in json_content
|
|
and "$ref" in json_content["schema"]
|
|
):
|
|
ref = json_content["schema"]["$ref"]
|
|
model_name = ref.split("/")[-1]
|
|
if model_name in schema["components"]["schemas"]:
|
|
model_schema = schema["components"]["schemas"][
|
|
model_name
|
|
]
|
|
if "example" in model_schema:
|
|
json_content["example"] = model_schema[
|
|
"example"
|
|
]
|
|
except KeyError:
|
|
pass
|
|
|
|
def configure_route_security(
|
|
self, path: str, method: str, schema: Dict[str, Any]
|
|
) -> None:
|
|
"""
|
|
Configure security requirements for a specific route.
|
|
|
|
Args:
|
|
path: Route path
|
|
method: HTTP method
|
|
schema: OpenAPI schema to modify
|
|
"""
|
|
# Check if route is protected based on dynamic routing info
|
|
if path in self.protected_routes and method in self.protected_routes[path]:
|
|
schema["paths"][path][method]["security"] = [
|
|
{"Bearer Auth": []},
|
|
]
|
|
schema["paths"][path][method]["responses"].update(
|
|
self._create_common_responses()
|
|
)
|
|
|
|
# Process request body examples
|
|
self._process_request_body(path, method, schema)
|
|
# Process response examples
|
|
self._process_response_examples(path, method, schema)
|
|
|
|
def create_schema(self) -> Dict[str, Any]:
|
|
"""
|
|
Create the complete OpenAPI schema.
|
|
|
|
Returns:
|
|
Dict[str, Any]: Complete OpenAPI schema
|
|
"""
|
|
openapi_schema = get_openapi(
|
|
title=Config.TITLE,
|
|
description=Config.DESCRIPTION,
|
|
version="1.1.1",
|
|
routes=self.app.routes,
|
|
tags=self.tags_metadata,
|
|
)
|
|
|
|
# Add security schemes
|
|
if "components" not in openapi_schema:
|
|
openapi_schema["components"] = {}
|
|
|
|
openapi_schema["components"][
|
|
"securitySchemes"
|
|
] = self._create_security_schemes()
|
|
# Configure route security and responses
|
|
for route in self.app.routes:
|
|
if isinstance(route, APIRoute) and route.include_in_schema:
|
|
path = str(route.path)
|
|
methods = [method.lower() for method in route.methods]
|
|
for method in methods:
|
|
self.configure_route_security(path, method, openapi_schema)
|
|
|
|
# # Add custom documentation extensions
|
|
openapi_schema["x-documentation"] = {
|
|
"postman_collection": "/docs/postman",
|
|
"swagger_ui": "/docs",
|
|
"redoc": "/redoc",
|
|
}
|
|
|
|
return openapi_schema
|
|
|
|
|
|
def create_openapi_schema(app: FastAPI) -> Dict[str, Any]:
|
|
"""
|
|
Create OpenAPI schema for a FastAPI application.
|
|
|
|
Args:
|
|
app: FastAPI application instance
|
|
|
|
Returns:
|
|
Dict[str, Any]: Complete OpenAPI schema
|
|
"""
|
|
creator = OpenAPISchemaCreator(app)
|
|
return creator.create_schema()
|