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.