FastAPI Boilerplate

Features

Async SQLAlchemy session
Custom user class
Dependencies for specific permissions
Celery
Dockerize(Hot reload)
Event dispatcher
Cache

Run

Launch docker

> docker-compose -f docker/docker-compose.yml up

Install dependency

> poetry shell
> poetry install

Apply alembic revision

> alembic upgrade head

Run server

> python3 main.py --env local|dev|prod --debug

Run test codes

> make test

Make coverage report

> make cov

Formatting

> pre-commit

SQLAlchemy for asyncio context

from core.db import Transactional, session


@Transactional()
async def create_user(self):
    session.add(User(email="padocon@naver.com"))

Do not use explicit commit(). Transactional class automatically do.

Query with asyncio.gather()

When executing queries concurrently through asyncio.gather(), you must use the session_factory context manager rather than the globally used session.

from core.db import session_factory


async def get_by_id(self, *, user_id) -> User:
    stmt = select(User)
    async with session_factory() as read_session:
        return await read_session.execute(query).scalars().first()


async def main() -> None:
    user_1, user_2 = await asyncio.gather(
        get_by_id(user_id=1),
        get_by_id(user_id=2),
    )

If you do not use a database connection like session.add(), it is recommended to use a globally provided session.

Multiple databases

Go to core/config.py and edit WRITER_DB_URL and READER_DB_URL in the config class.

If you need additional logic to use the database, refer to the get_bind() method of RoutingClass.

Custom user for authentication

from fastapi import Request


@home_router.get("/")
def home(request: Request):
    return request.user.id

Note. you have to pass jwt token via header like Authorization: Bearer 1234

Custom user class automatically decodes header token and store user information into request.user

If you want to modify custom user class, you have to update below files.

core/fastapi/schemas/current_user.py
core/fastapi/middlewares/authentication.py

CurrentUser

class CurrentUser(BaseModel):
    id: int = Field(None, description="ID")

Simply add more fields based on your needs.

AuthBackend

current_user = CurrentUser()

After line 18, assign values that you added on CurrentUser.

Top-level dependency

Note. Available from version 0.62 or higher.

Set a callable function when initialize FastAPI() app through dependencies argument.

Refer Logging class inside of core/fastapi/dependencies/logging.py

Dependencies for specific permissions

Permissions IsAdmin, IsAuthenticated, AllowAll have already been implemented.

from core.fastapi.dependencies import (
    PermissionDependency,
    IsAdmin,
)


user_router = APIRouter()


@user_router.get(
    "",
    response_model=List[GetUserListResponseSchema],
    response_model_exclude={"id"},
    responses={"400": {"model": ExceptionResponseSchema}},
    dependencies=[Depends(PermissionDependency([IsAdmin]))],  # HERE
)
async def get_user_list(
    limit: int = Query(10, description="Limit"),
    prev: int = Query(None, description="Prev ID"),
):
    pass

Insert permission through dependencies argument.

If you want to make your own permission, inherit BasePermission and implement has_permission() function.

Note. In order to use swagger's authorize function, you must put PermissionDependency as an argument of dependencies.

Event dispatcher

Refer the README of https://github.com/teamhide/fastapi-event

Cache

Caching by prefix

from core.helpers.cache import Cache


@Cache.cached(prefix="get_user", ttl=60)
async def get_user():
    ...

Caching by tag

from core.helpers.cache import Cache, CacheTag


@Cache.cached(tag=CacheTag.GET_USER_LIST, ttl=60)
async def get_user():
    ...

Use the Cache decorator to cache the return value of a function.

Depending on the argument of the function, caching is stored with a different value through internal processing.

Custom Key builder

from core.helpers.cache.base import BaseKeyMaker


class CustomKeyMaker(BaseKeyMaker):
    async def make(self, function: Callable, prefix: str) -> str:
        ...

If you want to create a custom key, inherit the BaseKeyMaker class and implement the make() method.

Custom Backend

from core.helpers.cache.base import BaseBackend


class RedisBackend(BaseBackend):
    async def get(self, key: str) -> Any:
        ...

    async def set(self, response: Any, key: str, ttl: int = 60) -> None:
        ...

    async def delete_startswith(self, value: str) -> None:
        ...

If you want to create a custom key, inherit the BaseBackend class and implement the get(), set(), delete_startswith() method.

Pass your custom backend or keymaker as an argument to init. (/app/server.py)

def init_cache() -> None:
    Cache.init(backend=RedisBackend(), key_maker=CustomKeyMaker())

Remove all cache by prefix/tag

from core.helpers.cache import Cache, CacheTag


await Cache.remove_by_prefix(prefix="get_user_list")
await Cache.remove_by_tag(tag=CacheTag.GET_USER_LIST)

teamhide/fastapi-boilerplate