baldur.interfaces — Web Framework
The web-framework abstraction (request/response context, method/content enums, permission levels, the handler type alias) plus the framework exception hierarchy. Adapter authors implement these to integrate Django, FastAPI, and Flask.
Enums
HttpMethod
Bases: str, Enum
HTTP methods
ContentType
Bases: str, Enum
Common content types
PermissionLevel
Bases: str, Enum
Framework-independent permission levels.
Maps to framework-specific permission classes in each adapter. Covers 98% of RBAC declarations (155/158). The remaining 1 special permission (ChaosTest) uses custom_permissions parameter.
DTOs
RequestContext
dataclass
RequestContext(
method: HttpMethod,
path: str,
headers: dict[str, str] = dict(),
query_params: dict[str, Any] = dict(),
path_params: dict[str, Any] = dict(),
body: bytes | None = None,
json_body: dict | None = None,
user: Any | None = None,
is_authenticated: bool = False,
client_ip: str | None = None,
user_agent: str | None = None,
request_id: str | None = None,
content_type: str | None = None,
)
Framework-independent request context.
Adapters convert framework-specific requests to this format, enabling handlers to work with any web framework.
Attributes:
| Name | Type | Description |
|---|---|---|
method |
HttpMethod
|
HTTP method |
path |
str
|
Request path (without query string) |
headers |
dict[str, str]
|
Request headers (case-insensitive keys) |
query_params |
dict[str, Any]
|
Query string parameters |
path_params |
dict[str, Any]
|
URL path parameters (e.g., {id}) |
body |
bytes | None
|
Raw request body bytes |
json_body |
dict | None
|
Parsed JSON body (if applicable) |
user |
Any | None
|
Authenticated user object (framework-specific) |
is_authenticated |
bool
|
Whether user is authenticated |
client_ip |
str | None
|
Client IP address |
user_agent |
str | None
|
User-Agent header value |
request_id |
str | None
|
Unique request identifier for tracing |
content_type |
str | None
|
Content-Type header value |
is_json
property
is_json: bool
Check if request has JSON content.
get_header
get_header(
name: str, default: str | None = None
) -> str | None
Get header value (case-insensitive).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Header name |
required |
default
|
str | None
|
Default value if not found |
None
|
Returns:
| Type | Description |
|---|---|
str | None
|
Header value or default |
get_query
get_query(name: str, default: Any = None) -> Any
Get query parameter value.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Parameter name |
required |
default
|
Any
|
Default value if not found |
None
|
Returns:
| Type | Description |
|---|---|
Any
|
Parameter value or default |
get_path_param
get_path_param(name: str, default: Any = None) -> Any
Get path parameter value.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Parameter name |
required |
default
|
Any
|
Default value if not found |
None
|
Returns:
| Type | Description |
|---|---|
Any
|
Parameter value or default |
ResponseContext
dataclass
ResponseContext(
status_code: int = 200,
body: Any = None,
headers: dict[str, str] = dict(),
content_type: str = ContentType.JSON.value,
is_streaming: bool = False,
)
Framework-independent response context.
Handlers return this, adapters convert to framework responses.
Attributes:
| Name | Type | Description |
|---|---|---|
status_code |
int
|
HTTP status code |
body |
Any
|
Response body (will be JSON-encoded if dict/list) |
headers |
dict[str, str]
|
Response headers |
content_type |
str
|
Content-Type header |
json
classmethod
json(
data: Any,
status_code: int = 200,
headers: dict[str, str] | None = None,
) -> ResponseContext
Create JSON response.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
Any
|
Data to JSON-encode |
required |
status_code
|
int
|
HTTP status code |
200
|
headers
|
dict[str, str] | None
|
Additional headers |
None
|
Returns:
| Type | Description |
|---|---|
ResponseContext
|
ResponseContext with JSON content |
error
classmethod
error(
message: str,
status_code: int = 400,
error_code: str | None = None,
details: dict | None = None,
) -> ResponseContext
Create error response.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message
|
str
|
Error message |
required |
status_code
|
int
|
HTTP status code |
400
|
error_code
|
str | None
|
Machine-readable error code |
None
|
details
|
dict | None
|
Additional error details |
None
|
Returns:
| Type | Description |
|---|---|
ResponseContext
|
ResponseContext with error body |
success
classmethod
success(
data: Any = None,
message: str | None = None,
status_code: int = 200,
) -> ResponseContext
Create success response.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
Any
|
Response data |
None
|
message
|
str | None
|
Success message |
None
|
status_code
|
int
|
HTTP status code |
200
|
Returns:
| Type | Description |
|---|---|
ResponseContext
|
ResponseContext with success body |
created
classmethod
created(
data: Any, location: str | None = None
) -> ResponseContext
Create 201 Created response.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
Any
|
Created resource data |
required |
location
|
str | None
|
Location header value (URL of new resource) |
None
|
Returns:
| Type | Description |
|---|---|
ResponseContext
|
ResponseContext with 201 status |
no_content
classmethod
no_content() -> ResponseContext
Create 204 No Content response.
Returns:
| Type | Description |
|---|---|
ResponseContext
|
ResponseContext with 204 status and no body |
not_found
classmethod
not_found(
message: str = "Resource not found",
) -> ResponseContext
Create 404 Not Found response.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message
|
str
|
Error message |
'Resource not found'
|
Returns:
| Type | Description |
|---|---|
ResponseContext
|
ResponseContext with 404 status |
unauthorized
classmethod
unauthorized(
message: str = "Authentication required",
) -> ResponseContext
Create 401 Unauthorized response.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message
|
str
|
Error message |
'Authentication required'
|
Returns:
| Type | Description |
|---|---|
ResponseContext
|
ResponseContext with 401 status |
forbidden
classmethod
forbidden(
message: str = "Permission denied",
) -> ResponseContext
Create 403 Forbidden response.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message
|
str
|
Error message |
'Permission denied'
|
Returns:
| Type | Description |
|---|---|
ResponseContext
|
ResponseContext with 403 status |
bad_request
classmethod
bad_request(
message: str = "Bad request", errors: dict | None = None
) -> ResponseContext
Create 400 Bad Request response.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message
|
str
|
Error message |
'Bad request'
|
errors
|
dict | None
|
Validation errors dict |
None
|
Returns:
| Type | Description |
|---|---|
ResponseContext
|
ResponseContext with 400 status |
server_error
classmethod
server_error(
message: str = "Internal server error",
) -> ResponseContext
Create 500 Internal Server Error response.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message
|
str
|
Error message |
'Internal server error'
|
Returns:
| Type | Description |
|---|---|
ResponseContext
|
ResponseContext with 500 status |
redirect
classmethod
redirect(
url: str, permanent: bool = False
) -> ResponseContext
Create redirect response.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
url
|
str
|
Redirect URL |
required |
permanent
|
bool
|
If True, use 301, else 302 |
False
|
Returns:
| Type | Description |
|---|---|
ResponseContext
|
ResponseContext with redirect |
streaming
classmethod
streaming(
body_iterator: Iterator[str | bytes],
content_type: str,
filename: str | None = None,
status_code: int = 200,
headers: dict[str, str] | None = None,
) -> ResponseContext
Framework-agnostic streaming response.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
body_iterator
|
Iterator[str | bytes]
|
Iterator yielding chunks of data |
required |
content_type
|
str
|
MIME type of the response |
required |
filename
|
str | None
|
Optional filename for Content-Disposition header |
None
|
status_code
|
int
|
HTTP status code |
200
|
headers
|
dict[str, str] | None
|
Additional response headers |
None
|
raw
classmethod
raw(
body: str | bytes,
content_type: str,
status_code: int = 200,
headers: dict[str, str] | None = None,
) -> ResponseContext
Non-JSON raw response (Prometheus text, plain text, etc.).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
body
|
str | bytes
|
Raw response body |
required |
content_type
|
str
|
MIME type of the response |
required |
status_code
|
int
|
HTTP status code |
200
|
headers
|
dict[str, str] | None
|
Additional response headers |
None
|
service_unavailable
classmethod
service_unavailable(
message: str = "Service not available",
error_code: str | None = None,
) -> ResponseContext
503 Service Unavailable convenience method.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message
|
str
|
Error message |
'Service not available'
|
error_code
|
str | None
|
Machine-readable error code |
None
|
Exceptions
WebFrameworkError
WebFrameworkError(message: str = '', *, code: str = '')
RouteNotFoundError
RouteNotFoundError(message: str = '', *, code: str = '')
AuthenticationError
AuthenticationError(message: str = '', *, code: str = '')
PermissionDeniedError
PermissionDeniedError(message: str = '', *, code: str = '')
Interface & types
WebFrameworkInterface
Bases: ABC
Abstract interface for web framework integration.
NOTE: No production implementation exists yet. Django is used directly. This interface is preserved as a design contract for future framework migration (Django -> FastAPI, Flask, etc.).
When implementing: - See adapters/django/ for reference patterns - Register via ProviderRegistry.register_web_framework()
Example
framework = ProviderRegistry.get_framework() router = framework.create_router(prefix="/api/v1")
def handle_status(request: RequestContext) -> ResponseContext: ... return ResponseContext.json({"status": "healthy"})
framework.add_route( ... router=router, ... path="/health", ... method=HttpMethod.GET, ... handler=handle_status, ... permission_level=PermissionLevel.PUBLIC, ... )
framework_name
abstractmethod
property
framework_name: str
Return the framework name.
Returns:
| Type | Description |
|---|---|
str
|
Framework identifier (e.g., 'django', 'fastapi', 'flask') |
create_router
abstractmethod
create_router(
prefix: str = "", tags: list[str] | None = None
) -> Any
Create a router/blueprint for grouping routes.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
prefix
|
str
|
URL prefix for all routes in this router |
''
|
tags
|
list[str] | None
|
OpenAPI tags for documentation |
None
|
Returns:
| Type | Description |
|---|---|
Any
|
Framework-specific router object |
Example
router = framework.create_router( ... prefix="/api/v1/payments", ... tags=["payments"], ... )
add_route
abstractmethod
add_route(
router: Any,
path: str,
method: HttpMethod,
handler: HandlerFunc,
permission_level: PermissionLevel = PermissionLevel.AUTHENTICATED,
custom_permissions: list[str] | None = None,
response_model: type | None = None,
summary: str | None = None,
description: str | None = None,
deprecated: bool = False,
) -> None
Add a route to the router.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
router
|
Any
|
Router from create_router |
required |
path
|
str
|
URL path (can include path parameters like {id}) |
required |
method
|
HttpMethod
|
HTTP method |
required |
handler
|
HandlerFunc
|
Handler function (RequestContext -> ResponseContext) |
required |
permission_level
|
PermissionLevel
|
Framework-independent permission level |
AUTHENTICATED
|
custom_permissions
|
list[str] | None
|
Special permission codes (1 edge case) |
None
|
response_model
|
type | None
|
Pydantic/Serializer model for response |
None
|
summary
|
str | None
|
OpenAPI summary (short description) |
None
|
description
|
str | None
|
OpenAPI description (detailed) |
None
|
deprecated
|
bool
|
Mark route as deprecated |
False
|
Example
framework.add_route( ... router=router, ... path="/{id}", ... method=HttpMethod.GET, ... handler=get_payment, ... summary="Get payment by ID", ... permission_level=PermissionLevel.VIEWER, ... )
include_router
abstractmethod
include_router(
parent: Any, child: Any, prefix: str = ""
) -> None
Include a child router in parent.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
parent
|
Any
|
Parent router or application |
required |
child
|
Any
|
Child router to include |
required |
prefix
|
str
|
Additional URL prefix |
''
|
Example
app = framework.create_app() framework.include_router(app, payment_router, prefix="/payments")
to_request_context
abstractmethod
to_request_context(request: Any) -> RequestContext
Convert framework request to RequestContext.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request
|
Any
|
Framework-specific request object |
required |
Returns:
| Type | Description |
|---|---|
RequestContext
|
Normalized RequestContext |
from_response_context
abstractmethod
from_response_context(response: ResponseContext) -> Any
Convert ResponseContext to framework response.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
response
|
ResponseContext
|
Framework-independent ResponseContext |
required |
Returns:
| Type | Description |
|---|---|
Any
|
Framework-specific response object |
wrap_handler
wrap_handler(handler: HandlerFunc) -> Callable
Wrap a handler function for the framework.
Converts between framework request/response and our contexts.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
handler
|
HandlerFunc
|
Handler using RequestContext/ResponseContext |
required |
Returns:
| Type | Description |
|---|---|
Callable
|
Framework-compatible handler function |
add_middleware
abstractmethod
add_middleware(
app: Any, middleware_class: type, **options: Any
) -> None
Add middleware to application.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
app
|
Any
|
Application instance |
required |
middleware_class
|
type
|
Middleware class |
required |
**options
|
Any
|
Middleware configuration options |
{}
|
Example
framework.add_middleware( ... app, ... CORSMiddleware, ... allow_origins=["*"], ... )
add_exception_handler
add_exception_handler(
app: Any,
exception_class: type[Exception],
handler: Callable[[Any, Exception], ResponseContext],
) -> None
Add exception handler to application.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
app
|
Any
|
Application instance |
required |
exception_class
|
type[Exception]
|
Exception type to handle |
required |
handler
|
Callable[[Any, Exception], ResponseContext]
|
Handler function (request, exception) -> ResponseContext |
required |
get_current_user
abstractmethod
get_current_user(request: Any) -> Any | None
Get authenticated user from request.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request
|
Any
|
Framework-specific request |
required |
Returns:
| Type | Description |
|---|---|
Any | None
|
User object or None if not authenticated |
require_auth
abstractmethod
require_auth() -> Callable
Get authentication dependency/decorator.
Returns:
| Type | Description |
|---|---|
Callable
|
Callable that enforces authentication |
Usage depends on framework
- Django: Permission class
- FastAPI: Dependency
- Flask: Decorator
require_permissions
abstractmethod
require_permissions(permissions: list[str]) -> Callable
Get permission checking dependency/decorator.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
permissions
|
list[str]
|
Required permission codes |
required |
Returns:
| Type | Description |
|---|---|
Callable
|
Callable that enforces permissions |
check_permission
check_permission(user: Any, permission: str) -> bool
Check if user has a specific permission.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
user
|
Any
|
User object |
required |
permission
|
str
|
Permission code to check |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if user has permission |
get_routes
get_routes(app: Any) -> list[dict]
Get list of registered routes.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
app
|
Any
|
Application instance |
required |
Returns:
| Type | Description |
|---|---|
list[dict]
|
List of route info dicts with path, method, name |
create_app
create_app(**options: Any) -> Any
Create a new application instance.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
**options
|
Any
|
Framework-specific options |
{}
|
Returns:
| Type | Description |
|---|---|
Any
|
Application instance |
run_server
run_server(
app: Any,
host: str = "0.0.0.0",
port: int = 8000,
debug: bool = False,
) -> None
Run development server.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
app
|
Any
|
Application instance |
required |
host
|
str
|
Host to bind |
'0.0.0.0'
|
port
|
int
|
Port to listen on |
8000
|
debug
|
bool
|
Enable debug mode |
False
|
Note
For development only. Use proper WSGI/ASGI server in production.
HandlerFunc
module-attribute
HandlerFunc = Callable[[RequestContext], ResponseContext]