fastack.middleware.init¶
MiddlewareManager
¶
Middleware Manager which allows you to create middleware with functions.
Source code in fastack/middleware/__init__.py
class MiddlewareManager:
"""
Middleware Manager which allows you to create middleware with functions.
"""
def __init__(self, app: "Fastack"):
self.app = app
def process_request(self, func: ProcessRequestFunc):
"""
Process request middleware.
Args:
func: Function to be called before request.
Example:
```python
from fastapi import Request
@app.middleware.process_request
def process_request(request):
print("Processing request")
auth = request.headers.get("Authorization")
if auth is None:
raise HTTPException(status_code=401)
```
"""
self.app.add_middleware(BaseMiddleware, process_request=func)
return func
def process_response(self, func: ProcessResponseFunc):
"""
Process response middleware.
Args:
func: Function to be called after request.
Example:
```python
from fastapi import Response
@app.middleware.process_response
def process_response(response: Response, exc: Exception = None):
print("Processing response")
success = "1" if exc is None else "0"
response.headers["X-Success"] = success
```
"""
self.app.add_middleware(BaseMiddleware, process_response=func)
return func
def process_websocket(self, func: ProcessWebSocketFunc):
"""
Process websocket middleware.
Args:
func: Function to be called before the request to WebSocket.
Example:
```python
from fastapi import WebSocket
@app.middleware.process_websocket
def process_websocket(websocket: WebSocket):
print("Processing websocket")
# websocket.accept()
```
"""
self.app.add_middleware(BaseMiddleware, process_websocket=func)
return func
def process_http(self, func: DispatchFunction):
"""
Original FastAPI.middleware
"""
self.app.add_middleware(BaseMiddleware, dispatch=func)
return func
def __call__(self, middleware_type: str) -> DecoratedMiddleware:
assert middleware_type in (
"http",
"request",
"response",
"websocket",
), "middleware_type must be 'http', 'request', 'response or 'websocket'"
func = getattr(self, "process_{}".format(middleware_type))
return func
process_http(self, func)
¶
process_request(self, func)
¶
Process request middleware.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
func |
Callable[[starlette.requests.Request], Awaitable[NoneType]] |
Function to be called before request. |
required |
Examples:
from fastapi import Request
@app.middleware.process_request
def process_request(request):
print("Processing request")
auth = request.headers.get("Authorization")
if auth is None:
raise HTTPException(status_code=401)
Source code in fastack/middleware/__init__.py
def process_request(self, func: ProcessRequestFunc):
"""
Process request middleware.
Args:
func: Function to be called before request.
Example:
```python
from fastapi import Request
@app.middleware.process_request
def process_request(request):
print("Processing request")
auth = request.headers.get("Authorization")
if auth is None:
raise HTTPException(status_code=401)
```
"""
self.app.add_middleware(BaseMiddleware, process_request=func)
return func
process_response(self, func)
¶
Process response middleware.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
func |
Callable[[starlette.responses.Response, Optional[Exception]], Awaitable[NoneType]] |
Function to be called after request. |
required |
Examples:
from fastapi import Response
@app.middleware.process_response
def process_response(response: Response, exc: Exception = None):
print("Processing response")
success = "1" if exc is None else "0"
response.headers["X-Success"] = success
Source code in fastack/middleware/__init__.py
def process_response(self, func: ProcessResponseFunc):
"""
Process response middleware.
Args:
func: Function to be called after request.
Example:
```python
from fastapi import Response
@app.middleware.process_response
def process_response(response: Response, exc: Exception = None):
print("Processing response")
success = "1" if exc is None else "0"
response.headers["X-Success"] = success
```
"""
self.app.add_middleware(BaseMiddleware, process_response=func)
return func
process_websocket(self, func)
¶
Process websocket middleware.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
func |
Callable[[starlette.websockets.WebSocket], Awaitable[NoneType]] |
Function to be called before the request to WebSocket. |
required |
Examples:
from fastapi import WebSocket
@app.middleware.process_websocket
def process_websocket(websocket: WebSocket):
print("Processing websocket")
# websocket.accept()
Source code in fastack/middleware/__init__.py
def process_websocket(self, func: ProcessWebSocketFunc):
"""
Process websocket middleware.
Args:
func: Function to be called before the request to WebSocket.
Example:
```python
from fastapi import WebSocket
@app.middleware.process_websocket
def process_websocket(websocket: WebSocket):
print("Processing websocket")
# websocket.accept()
```
"""
self.app.add_middleware(BaseMiddleware, process_websocket=func)
return func
Last update:
January 17, 2022 13:26:42