FastAPI Contrib
Opinionated set of utilities on top of FastAPI
- Free software: MIT license
- Documentation: https://fastapi-contrib.readthedocs.io.
Features
- Auth Backend & Middleware (User or None in every request object)
- Permissions: reusable class permissions, specify multiple as FastAPI Dependency
- ModelSerializers: serialize (pydantic) incoming request, connect data with DB model and save
- UJSONResponse: correctly show slashes in fields with URLs
- Limit-Offset Pagination: use it as FastAPI Dependency (works only with ModelSerializers for now)
- MongoDB integration: Use models as if it was Django (based on pydantic models)
- MongoDB indices verification on startup of the app
- Custom Exceptions and Custom Exception Handlers
- Opentracing middleware & setup utility with Jaeger tracer + root span available in every Request's state
- StateRequestIDMiddleware: receives configurable header and saves it in request state
Roadmap
See GitHub Project Roadmap.
Installation
To install just Contrib (without mongodb, pytz, ujson):
$ pip install fastapi_contrib
To install contrib with mongodb support:
$ pip install fastapi_contrib[mongo]
To install contrib with ujson support:
$ pip install fastapi_contrib[ujson]
To install contrib with pytz support:
$ pip install fastapi_contrib[pytz]
To install contrib with opentracing & Jaeger tracer:
$ pip install fastapi_contrib[jaegertracing]
To install everything:
$ pip install fastapi_contrib[all]
Usage
To use Limit-Offset pagination:
from fastapi import FastAPI
from fastapi_contrib.pagination import Pagination
from fastapi_contrib.serializers.common import ModelSerializer
from yourapp.models import SomeModel
app = FastAPI()
class SomeSerializer(ModelSerializer):
class Meta:
model = SomeModel
@app.get("/")
async def list(pagination: Pagination = Depends()):
filter_kwargs = {}
return await pagination.paginate(
serializer_class=SomeSerializer, **filter_kwargs
)
Subclass this pagination to define custom default & maximum values for offset & limit:
class CustomPagination(Pagination):
default_offset = 90
default_limit = 1
max_offset = 100
max_limit = 2000
To use State Request ID Middleware:
from fastapi import FastAPI
from fastapi_contrib.common.middlewares import StateRequestIDMiddleware
app = FastAPI()
@app.on_event('startup')
async def startup():
app.add_middleware(StateRequestIDMiddleware)
To use Authentication Middleware:
from fastapi import FastAPI
from fastapi_contrib.auth.backends import AuthBackend
from fastapi_contrib.auth.middlewares import AuthenticationMiddleware
app = FastAPI()
@app.on_event('startup')
async def startup():
app.add_middleware(AuthenticationMiddleware, backend=AuthBackend())
Define & use custom permissions based on FastAPI Dependency framework:
from fastapi import FastAPI
from fastapi_contrib.permissions import BasePermission, PermissionsDependency
class TeapotUserAgentPermission(BasePermission):
def has_required_permissions(self, request: Request) -> bool:
return request.headers.get('User-Agent') == "Teapot v1.0"
app = FastAPI()
@app.get(
"/teapot/",
dependencies=[Depends(
PermissionsDependency([TeapotUserAgentPermission]))]
)
async def teapot() -> dict:
return {"teapot": True}
Setup uniform exception-handling:
from fastapi import FastAPI
from fastapi_contrib.exception_handlers import setup_exception_handlers
app = FastAPI()
@app.on_event('startup')
async def startup():
setup_exception_handlers(app)
If you want to correctly handle scenario when request is an empty body (IMPORTANT: non-multipart):
from fastapi import FastAPI
from fastapi_contrib.routes import ValidationErrorLoggingRoute
app = FastAPI()
app.router.route_class = ValidationErrorLoggingRoute
Or if you use multiple routes for handling different namespaces (IMPORTANT: non-multipart):
from fastapi import APIRouter, FastAPI
from fastapi_contrib.routes import ValidationErrorLoggingRoute
app = FastAPI()
my_router = APIRouter(route_class=ValidationErrorLoggingRoute)
To correctly show slashes in fields with URLs + ascii locking:
from fastapi import FastAPI
from fastapi_contrib.common.responses import UJSONResponse
app = FastAPI()
@app.get("/", response_class=UJSONResponse)
async def root():
return {"a": "b"}
Or specify it as default response class for the whole app (FastAPI >= 0.39.0):
from fastapi import FastAPI
from fastapi_contrib.common.responses import UJSONResponse
app = FastAPI(default_response_class=UJSONResponse)
To setup Jaeger tracer and enable Middleware that captures every request in opentracing span:
from fastapi import FastAPI
from fastapi_contrib.tracing.middlewares import OpentracingMiddleware
from fastapi_contrib.tracing.utils import setup_opentracing
app = FastAPI()
@app.on_event('startup')
async def startup():
setup_opentracing(app)
app.add_middleware(OpentracingMiddleware)
To setup mongodb connection at startup and never worry about it again:
from fastapi import FastAPI
from fastapi_contrib.db.utils import setup_mongodb
app = FastAPI()
@app.on_event('startup')
async def startup():
setup_mongodb(app)
Use models to map data to MongoDB:
from fastapi_contrib.db.models import MongoDBModel
class MyModel(MongoDBModel):
additional_field1: str
optional_field2: int = 42
class Meta:
collection = "mymodel_collection"
mymodel = MyModel(additional_field1="value")
mymodel.save()
assert mymodel.additional_field1 == "value"
assert mymodel.optional_field2 == 42
assert isinstance(mymodel.id, int)
Or use TimeStamped model with creation datetime:
from fastapi_contrib.db.models import MongoDBTimeStampedModel
class MyTimeStampedModel(MongoDBTimeStampedModel):
class Meta:
collection = "timestamped_collection"
mymodel = MyTimeStampedModel()
mymodel.save()
assert isinstance(mymodel.id, int)
assert isinstance(mymodel.created, datetime)
Use serializers and their response models to correctly show Schemas and convert from JSON/dict to models and back:
from fastapi import FastAPI
from fastapi_contrib.db.models import MongoDBModel
from fastapi_contrib.serializers import openapi
from fastapi_contrib.serializers.common import Serializer
from yourapp.models import SomeModel
app = FastAPI()
class SomeModel(MongoDBModel):
field1: str
@openapi.patch
class SomeSerializer(Serializer):
read_only1: str = "const"
write_only2: int
not_visible: str = "42"
class Meta:
model = SomeModel
exclude = {"not_visible"}
write_only_fields = {"write_only2"}
read_only_fields = {"read_only1"}
@app.get("/", response_model=SomeSerializer.response_model)
async def root(serializer: SomeSerializer):
model_instance = await serializer.save()
return model_instance.dict()
POST-ing to this route following JSON:
{"read_only1": "a", "write_only2": 123, "field1": "b"}
Should return following response:
{"id": 1, "field1": "b", "read_only1": "const"}
Auto-creation of MongoDB indexes
Suppose we have this directory structure:
-- project_root/
-- apps/
-- app1/
-- models.py (with MongoDBModel inside with indices declared)
-- app2/
-- models.py (with MongoDBModel inside with indices declared)
Based on this, your name of the folder with all the apps would be "apps". This is the default name for fastapi_contrib package to pick up your structure automatically. You can change that by setting ENV variable CONTRIB_APPS_FOLDER_NAME (by the way, all the setting of this package are overridable via ENV vars with CONTRIB_ prefix before them).
You also need to tell fastapi_contrib which apps to look into for your models. This is controlled by CONTRIB_APPS ENV variable, which is list of str names of the apps with models. In the example above, this would be CONTRIB_APPS=["app1","app2"].
Just use create_indexes function after setting up mongodb:
from fastapi import FastAPI
from fastapi_contrib.db.utils import setup_mongodb, create_indexes
app = FastAPI()
@app.on_event("startup")
async def startup():
setup_mongodb(app)
await create_indexes()
This will scan all the specified CONTRIB_APPS in the CONTRIB_APPS_FOLDER_NAME for models, that are subclassed from either MongoDBModel or MongoDBTimeStampedModel and create indices for any of them that has Meta class with indexes attribute:
models.py:
import pymongo
from fastapi_contrib.db.models import MongoDBTimeStampedModel
class MyModel(MongoDBTimeStampedModel):
class Meta:
collection = "mymodel"
indexes = [
pymongo.IndexModel(...),
pymongo.IndexModel(...),
]
This would not create duplicate indices because it relies on pymongo and motor to do all the job.
Credits
This package was created with Cookiecutter and the audreyr/cookiecutter-pypackage project template.
871 Dec 25, 2022
153 Oct 21, 2022
562 Jan 01, 2023
441 Jan 05, 2023
144 Dec 28, 2022
4 Dec 05, 2022
558 Jan 07, 2023
12 Jan 07, 2023
291 Jan 06, 2023
225 Dec 28, 2022
1.3k Jan 04, 2023
27 Dec 19, 2022
18 Jul 04, 2022
9 Nov 15, 2022
0 Sep 24, 2022
3 May 09, 2022
131 Sep 26, 2022
2 Aug 15, 2022
1 Nov 28, 2021
14 Jan 06, 2022