📖 Documentation

0. About

🚀 Join Our Community

1. Features

2. Contents

3. Prerequisites

3.0 Start

git clone https://github.com/igormagalhaesr/FastAPI-boilerplate

Tip

If you are in a hurry, you may use one of the following templates (containing a.env,docker-compose.yml andDockerfile):

Warning

Do not forget to placedocker-compose.yml andDockerfile in theroot folder, while.env should be in thesrc folder.

3.1 Environment Variables (.env)

touch .env

# ------------- app settings -------------APP_NAME="Your app name here"APP_DESCRIPTION="Your app description here"APP_VERSION="0.1"CONTACT_NAME="Your name"CONTACT_EMAIL="Your email"LICENSE_NAME="The license you picked"

# ------------- database -------------POSTGRES_USER="your_postgres_user"POSTGRES_PASSWORD="your_password"POSTGRES_SERVER="your_server" # default "localhost", if using docker compose you should use "db"POSTGRES_PORT=5432 # default "5432", if using docker compose you should use "5432"POSTGRES_DB="your_db"

# ------------- pgadmin -------------PGADMIN_DEFAULT_EMAIL="your_email_address"PGADMIN_DEFAULT_PASSWORD="your_password"PGADMIN_LISTEN_PORT=80

openssl rand -hex 32

# ------------- crypt -------------SECRET_KEY= # result of openssl rand -hex 32ALGORITHM= # pick an algorithm, default HS256ACCESS_TOKEN_EXPIRE_MINUTES= # minutes until token expires, default 30REFRESH_TOKEN_EXPIRE_DAYS= # days until token expires, default 7

# ------------- admin -------------ADMIN_NAME="your_name"ADMIN_EMAIL="your_email"ADMIN_USERNAME="your_username"ADMIN_PASSWORD="your_password"

# ------------- crud admin -------------CRUD_ADMIN_ENABLED=true                    # default=true, set to false to disable admin panelCRUD_ADMIN_MOUNT_PATH="/admin"             # default="/admin", path where admin panel will be mounted# ------------- crud admin security -------------CRUD_ADMIN_MAX_SESSIONS=10                 # default=10, maximum concurrent sessions per userCRUD_ADMIN_SESSION_TIMEOUT=1440            # default=1440 (24 hours), session timeout in minutesSESSION_SECURE_COOKIES=true                # default=true, use secure cookies# ------------- crud admin tracking -------------CRUD_ADMIN_TRACK_EVENTS=true               # default=true, track admin eventsCRUD_ADMIN_TRACK_SESSIONS=true             # default=true, track admin sessions in database# ------------- crud admin redis (optional for production) -------------CRUD_ADMIN_REDIS_ENABLED=false             # default=false, use Redis for session storageCRUD_ADMIN_REDIS_HOST="localhost"          # default="localhost", Redis host for admin sessionsCRUD_ADMIN_REDIS_PORT=6379                 # default=6379, Redis port for admin sessionsCRUD_ADMIN_REDIS_DB=0                      # default=0, Redis database for admin sessionsCRUD_ADMIN_REDIS_PASSWORD=""               # optional, Redis password for admin sessionsCRUD_ADMIN_REDIS_SSL=false                 # default=false, use SSL for Redis connection

# ------------- redis cache-------------REDIS_CACHE_HOST="your_host" # default "localhost", if using docker compose you should use "redis"REDIS_CACHE_PORT=6379 # default "6379", if using docker compose you should use "6379"

# ------------- redis client-side cache -------------CLIENT_CACHE_MAX_AGE=30 # default "30"

# ------------- redis queue -------------REDIS_QUEUE_HOST="your_host" # default "localhost", if using docker compose you should use "redis"REDIS_QUEUE_PORT=6379 # default "6379", if using docker compose you should use "6379"

Warning

You may use the same redis for both caching and queue while developing, but the recommendation is using two separate containers for production.

# ------------- first tier -------------TIER_NAME="free"

# ------------- redis rate limit -------------REDIS_RATE_LIMIT_HOST="localhost"   # default="localhost", if using docker compose you should use "redis"REDIS_RATE_LIMIT_PORT=6379          # default=6379, if using docker compose you should use "6379"# ------------- default rate limit settings -------------DEFAULT_RATE_LIMIT_LIMIT=10         # default=10DEFAULT_RATE_LIMIT_PERIOD=3600      # default=3600

# ------------- environment -------------ENVIRONMENT="local"

3.2 Docker Compose (preferred)

docker compose up

3.3 From Scratch

pip install uv

4. Usage

4.1 Docker Compose

docker compose up

fastapi-boilerplate-worker-1| ... redis_version=x.x.x mem_usage=999K clients_connected=1 db_keys=0...fastapi-boilerplate-db-1| ... [1] LOG:  database system is ready to accept connections...fastapi-boilerplate-web-1| INFO:     Application startup complete.

4.2 From Scratch

4.2.1. Packages

uv sync

4.2.2. Running PostgreSQL With Docker

Note

If you already have a PostgreSQL running, you may skip this step.

docker pull postgres

docker run -d \    -p {PORT}:{PORT} \    --name {NAME} \    -e POSTGRES_PASSWORD={PASSWORD} \    -e POSTGRES_USER={USER} \    postgres

docker run -d \    -p 5432:5432 \    --name postgres \    -e POSTGRES_PASSWORD=1234 \    -e POSTGRES_USER=postgres \    postgres

4.2.3. Running redis With Docker

Note

If you already have a redis running, you may skip this step.

docker pull redis:alpine

docker run -d \  --name {NAME}  \  -p {PORT}:{PORT} \redis:alpine

docker run -d \  --name redis  \  -p 6379:6379 \redis:alpine

4.2.4. Running the API

uv run uvicorn src.app.main:app --reload

Tip

The --reload flag enables auto-reload once you change (and save) something in the project

4.3 Creating the first superuser

4.3.1 Docker Compose

Warning

Make sure DB and tables are created before running create_superuser (db should be running and the api should run at least once before)

  #-------- uncomment to create first superuser --------  # create_superuser:  #   build:  #     context: .  #     dockerfile: Dockerfile  #   env_file:  #     - ./src/.env  #   depends_on:  #     - db  #   command: python -m src.scripts.create_first_superuser  #   volumes:  #     - ./src:/code/src

  #-------- uncomment to create first superuser --------  create_superuser:    build:      context: .      dockerfile: Dockerfile    env_file:      - ./src/.env    depends_on:      - db    command: python -m src.scripts.create_first_superuser    volumes:      - ./src:/code/src

docker-compose up -d

docker-compose run --rm create_superuser

docker-compose stop create_superuser

4.3.2 From Scratch

uv run python -m src.scripts.create_first_superuser

4.3.3 Creating the first tier

Warning

Make sure DB and tables are created before running create_tier (db should be running and the api should run at least once before)

4.4 Database Migrations

Warning

To create the tables if you did not create the endpoints, ensure that you import the models in src/app/models/init.py. This step is crucial to create the new tables.

  db:    image: postgres:13    env_file:      - ./src/.env    volumes:      - postgres-data:/var/lib/postgresql/data# -------- replace with comment to run migrations with docker --------    expose:      -"5432"# ports:#  - 5432:5432

  db:    ...# expose:#  - "5432"    ports:      - 5432:5432

uv run alembic revision --autogenerate

uv run alembic upgrade head

Note

If you do not have uv, you may run it without uv after runningpip install alembic

5. Extending

5.1 Project Structure

.├── Dockerfile# Dockerfile for building the application container.├── docker-compose.yml# Docker Compose file for defining multi-container applications.├── pyproject.toml# Project configuration file with metadata and dependencies (PEP 621).├── uv.lock# uv lock file specifying exact versions of dependencies.├── README.md# Project README providing information and instructions.├── LICENSE.md# License file for the project.│├── tests# Unit tests for the application.│   ├──helpers# Helper functions for tests.│   │   ├── generators.py# Helper functions for generating test data.│   │   └── mocks.py# Mock functions for testing.│   ├── __init__.py│   ├── conftest.py# Configuration and fixtures for pytest.│   └── test_user_unit.py# Unit test cases for user-related functionality.│└── src# Source code directory.    ├── __init__.py# Initialization file for the src package.    ├── alembic.ini# Configuration file for Alembic (database migration tool).    │    ├── app# Main application directory.    │   ├── __init__.py# Initialization file for the app package.    │   ├── main.py# Main entry point of the FastAPI application.    │   │    │   │    │   ├── api# Folder containing API-related logic.    │   │   ├── __init__.py    │   │   ├── dependencies.py# Defines dependencies for use across API endpoints.    │   │   │    │   │   └── v1# Version 1 of the API.    │   │       ├── __init__.py    │   │       ├── login.py# API route for user login.    │   │       ├── logout.py# API route for user logout.    │   │       ├── posts.py# API routes for post operations.    │   │       ├── rate_limits.py# API routes for rate limiting functionalities.    │   │       ├── tasks.py# API routes for task management.    │   │       ├── tiers.py# API routes for user tier functionalities.    │   │       └── users.py# API routes for user management.    │   │    │   ├── core# Core utilities and configurations for the application.    │   │   ├── __init__.py    │   │   ├── config.py# Configuration settings for the application.    │   │   ├── logger.py# Configuration for application logging.    │   │   ├── schemas.py# Pydantic schemas for data validation.    │   │   ├── security.py# Security utilities, such as password hashing.    │   │   ├── setup.py# Setup file for the FastAPI app instance.    │   │   │    │   │   ├── db# Core Database related modules.    │   │   │   ├── __init__.py    │   │   │   ├── crud_token_blacklist.py# CRUD operations for token blacklist.    │   │   │   ├── database.py# Database connectivity and session management.    │   │   │   ├── models.py# Core Database models.    │   │   │   └── token_blacklist.py# Model for token blacklist functionality.    │   │   │    │   │   ├── exceptions# Custom exception classes.    │   │   │   ├── __init__.py    │   │   │   ├── cache_exceptions.py# Exceptions related to cache operations.    │   │   │   └── http_exceptions.py# HTTP-related exceptions.    │   │   │    │   │   ├── utils# Utility functions and helpers.    │   │   │   ├── __init__.py    │   │   │   ├── cache.py# Cache-related utilities.    │   │   │   ├── queue.py# Utilities for task queue management.    │   │   │   └── rate_limit.py# Rate limiting utilities.    │   │   │    │   │   └── worker# Worker script for background tasks.    │   │       ├── __init__.py    │   │       ├── settings.py# Worker configuration and settings.    │   │       └── functions.py# Async task definitions and management.    │   │    │   ├── crud# CRUD operations for the application.    │   │   ├── __init__.py    │   │   ├── crud_base.py# Base class for CRUD operations.    │   │   ├── crud_posts.py# CRUD operations for posts.    │   │   ├── crud_rate_limit.py# CRUD operations for rate limiting.    │   │   ├── crud_tier.py# CRUD operations for user tiers.    │   │   ├── crud_users.py# CRUD operations for users.    │   │   └── helper.py# Helper functions for CRUD operations.    │   │    │   ├── logs# Directory for log files.    │   │   └── app.log# Log file for the application.    │   │    │   ├── middleware# Middleware components for the application.    │   │   └── client_cache_middleware.py# Middleware for client-side caching.    │   │    │   ├── models# ORM models for the application.    │   │   ├── __init__.py    │   │   ├── post.py# ORM model for posts.    │   │   ├── rate_limit.py# ORM model for rate limiting.    │   │   ├── tier.py# ORM model for user tiers.    │   │   └── user.py# ORM model for users.    │   │    │   └── schemas# Pydantic schemas for data validation.    │       ├── __init__.py    │       ├── job.py# Schema for background jobs.    │       ├── post.py# Schema for post data.    │       ├── rate_limit.py# Schema for rate limiting data.    │       ├── tier.py# Schema for user tier data.    │       └── user.py# Schema for user data.    │    ├── migrations# Alembic migration scripts for database changes.    │   ├── README    │   ├── env.py# Environment configuration for Alembic.    │   ├── script.py.mako# Template script for Alembic migrations.    │   │    │   └── versions# Individual migration scripts.    │       └── README.MD    │    └── scripts# Utility scripts for the application.        ├── __init__.py        ├── create_first_superuser.py# Script to create the first superuser.        └── create_first_tier.py# Script to create the first user tier.

5.2 Database Model

5.2.1 Token Blacklist

5.3 SQLAlchemy Models

Warning

Note that since it inherits fromBase, the new model is mapped as a pythondataclass, so optional attributes (arguments with a default value) should be defined after required attributes.

fromsqlalchemyimportString,DateTimefromsqlalchemy.ormimportMapped,mapped_column,relationshipfromapp.core.db.databaseimportBaseclassEntity(Base):__tablename__="entity"id:Mapped[int]=mapped_column("id",autoincrement=True,nullable=False,unique=True,primary_key=True,init=False)name:Mapped[str]=mapped_column(String(30))    ...

5.4 Pydantic Schemas

fromtypingimportAnnotatedfrompydanticimportBaseModel,EmailStr,Field,HttpUrl,ConfigDictclassEntityBase(BaseModel):name:Annotated[str,Field(min_length=2,max_length=30,examples=["Entity Name"]),    ]classEntity(EntityBase):    ...classEntityRead(EntityBase):    ...classEntityCreate(EntityBase):    ...classEntityCreateInternal(EntityCreate):    ...classEntityUpdate(BaseModel):    ...classEntityUpdateInternal(BaseModel):    ...classEntityDelete(BaseModel):model_config=ConfigDict(extra="forbid")is_deleted:booldeleted_at:datetime

5.5 Alembic Migrations

Warning

To create the tables if you did not create the endpoints, ensure that you import the models in src/app/models/init.py. This step is crucial to create the new models.

uv run alembic revision --autogenerate

uv run alembic upgrade head

5.6 CRUD

fromfastcrudimportFastCRUDfromapp.models.entityimportEntityfromapp.schemas.entityimportEntityCreateInternal,EntityUpdate,EntityUpdateInternal,EntityDeleteCRUDEntity=FastCRUD[Entity,EntityCreateInternal,EntityUpdate,EntityUpdateInternal,EntityDelete]crud_entity=CRUDEntity(Entity)

# crud_users.pyfromapp.model.userimportUserfromapp.schemas.userimportUserCreateInternal,UserUpdate,UserUpdateInternal,UserDeleteCRUDUser=FastCRUD[User,UserCreateInternal,UserUpdate,UserUpdateInternal,UserDelete]crud_users=CRUDUser(User)

5.6.1 Get

# Here I'm getting the first user with email == user.email (email is unique in this case)user=awaitcrud_users.get(db=db,email=user.email)

5.6.2 Get Multi

# Here I'm getting at most 10 users with the name 'User Userson' except for the first 3user=awaitcrud_users.get_multi(db=db,offset=3,limit=100,name="User Userson")

Warning

Note that get_multi returns a pythondict.

{"data":[{"id":4,"name":"User Userson","username":"userson4","email":"user.userson4@example.com","profile_image_url":"https://profileimageurl.com"},{"id":5,"name":"User Userson","username":"userson5","email":"user.userson5@example.com","profile_image_url":"https://profileimageurl.com"}],"total_count":2,"has_more":false,"page":1,"items_per_page":10}

5.6.3 Create

fromapp.schemas.userimportUserCreate# Creating the objectuser_internal=UserCreate(name="user",username="myusername",email="user@example.com")# Passing the object to be createdcrud_users.create(db=db,object=user_internal)

5.6.4 Exists

# This queries only the email variable# It returns True if there's at least one or False if there is nonecrud_users.exists(db=db,email=user @example.com)

5.6.5 Count

# Here I'm getting the count of users with the name 'User Userson'user=awaitcrud_users.count(db=db,name="User Userson")

5.6.6 Update

# Here I'm updating the user with username == "myusername".# #I'll change his name to "Updated Name"crud_users.update(db=db,object={"name":"Updated Name"},username="myusername")

5.6.7 Delete

# Here I'll just change is_deleted to Truecrud_users.delete(db=db,username="myusername")# Here I actually delete it from the databasecrud_users.db_delete(db=db,username="myusername")

5.6.8 Get Joined

# Fetch a single record with a join on another model (e.g., User and Tier).result=awaitcrud_users.get_joined(db=db,# The SQLAlchemy async session.join_model=Tier,# The model to join with (e.g., Tier).schema_to_select=UserSchema,# Pydantic schema for selecting User model columns (optional).join_schema_to_select=TierSchema,# Pydantic schema for selecting Tier model columns (optional).)

5.6.9 Get Multi Joined

# Retrieve a list of objects with a join on another model (e.g., User and Tier).result=awaitcrud_users.get_multi_joined(db=db,# The SQLAlchemy async session.join_model=Tier,# The model to join with (e.g., Tier).join_prefix="tier_",# Optional prefix for joined model columns.join_on=and_(User.tier_id==Tier.id,User.is_superuser==True),# Custom join condition.schema_to_select=UserSchema,# Pydantic schema for selecting User model columns.join_schema_to_select=TierSchema,# Pydantic schema for selecting Tier model columns.username="john_doe",# Additional filter parameters.)

More Efficient Selecting

fromapp.schemas.userimportUserRead# Here it's selecting all of the user's datacrud_user.get(db=db,username="myusername")# Now it's only selecting the data that is in UserRead.# Since that's my response_model, it's all I needcrud_user.get(db=db,username="myusername",schema_to_select=UserRead)

5.7 Routes

fromtypingimportAnnotated,ListfromfastapiimportDepends,Request,APIRouterfromsqlalchemy.ext.asyncioimportAsyncSessionfromapp.schemas.entityimportEntityReadfromapp.core.db.databaseimportasync_get_dbfromapp.crud.crud_entityimportcrud_entityrouter=APIRouter(tags=["entities"])@router.get("/entities/{id}",response_model=EntityRead)asyncdefread_entity(request:Request,id:int,db:Annotated[AsyncSession,Depends(async_get_db)]):entity=awaitcrud_entity.get(db=db,id=id)ifentityisNone:# Explicit None checkraiseNotFoundException("Entity not found")returnentity@router.get("/entities",response_model=List[EntityRead])asyncdefread_entities(request:Request,db:Annotated[AsyncSession,Depends(async_get_db)]):entities=awaitcrud_entity.get_multi(db=db,is_deleted=False)returnentities

fromfastapiimportAPIRouterfromapp.api.v1.entitiesimportrouterasentity_routerfromapp.api.v1.usersimportrouterasuser_routerfromapp.api.v1.postsimportrouteraspost_routerrouter=APIRouter(prefix="/v1")router.include_router(user_router)router.include_router(post_router)router.include_router(entity_router)# Add your new router

5.7.1 Paginated Responses

{"data":[{"id":4,"name":"User Userson","username":"userson4","email":"user.userson4@example.com","profile_image_url":"https://profileimageurl.com"},{"id":5,"name":"User Userson","username":"userson5","email":"user.userson5@example.com","profile_image_url":"https://profileimageurl.com"}],"total_count":2,"has_more":false,"page":1,"items_per_page":10}

fromtypingimportAnnotatedfromfastapiimportDepends,Requestfromsqlalchemy.ext.asyncioimportAsyncSessionfromfastcrud.paginatedimport (PaginatedListResponse,# What you'll use as a response_model to validatepaginated_response,# Creates a paginated response based on the parameterscompute_offset,# Calculate the offset for pagination ((page - 1) * items_per_page))

importfastapifromapp.schemas.entityimportEntityRead...@router.get("/entities",response_model=PaginatedListResponse[EntityRead])asyncdefread_entities(request:Request,db:Annotated[AsyncSession,Depends(async_get_db)],page:int=1,items_per_page:int=10):entities_data=awaitcrud_entity.get_multi(db=db,offset=compute_offset(page,items_per_page),limit=items_per_page,schema_to_select=EntityRead,is_deleted=False,    )returnpaginated_response(crud_data=entities_data,page=page,items_per_page=items_per_page)

5.7.2 HTTP Exceptions

fromapp.core.exceptions.http_exceptionsimport (NotFoundException,ForbiddenException,DuplicateValueException)@router.post("/entities",response_model=EntityRead,status_code=201)asyncdefcreate_entity(request:Request,entity_data:EntityCreate,db:Annotated[AsyncSession,Depends(async_get_db)],current_user:Annotated[UserRead,Depends(get_current_user)]):# Check if entity already existsifawaitcrud_entity.exists(db=db,name=entity_data.name)isTrue:raiseDuplicateValueException("Entity with this name already exists")# Check user permissionsifcurrent_user.is_activeisFalse:# Explicit boolean checkraiseForbiddenException("User account is disabled")# Create the entityentity=awaitcrud_entity.create(db=db,object=entity_data)ifentityisNone:# Explicit None checkraiseCustomException("Failed to create entity")returnentity@router.get("/entities/{id}",response_model=EntityRead)asyncdefread_entity(request:Request,id:int,db:Annotated[AsyncSession,Depends(async_get_db)]):entity=awaitcrud_entity.get(db=db,id=id)ifentityisNone:# Explicit None checkraiseNotFoundException("Entity not found")returnentity

5.8 Caching

Warning

Note that you should always pass request as a variable to your endpoint function if you plan to use the cache decorator.

...fromapp.core.utils.cacheimportcache@app.get("/sample/{my_id}")@cache(key_prefix="sample_data",expiration=3600,resource_id_name="my_id")asyncdefsample_endpoint(request:Request,my_id:int):# Endpoint logic herereturn {"data":"my_data"}

...fromapp.core.utils.cacheimportcache@app.get("/sample/{my_id}")@cache(key_prefix="sample_data",resource_id_type=int)asyncdefsample_endpoint(request:Request,my_id:int):# Endpoint logic herereturn {"data":"my_data"}

5.9 More Advanced Caching

Invalidating Extra Keys

# The cache here will be saved as "{username}_posts:{username}":@router.get("/{username}/posts",response_model=List[PostRead])@cache(key_prefix="{username}_posts",resource_id_name="username")asyncdefread_posts(request:Request,username:str,db:Annotated[AsyncSession,Depends(async_get_db)]):    ......# Invalidating cache for the former endpoint by just passing the key_prefix and id as a dictionary:@router.delete("/{username}/post/{id}")@cache("{username}_post_cache",resource_id_name="id",to_invalidate_extra={"{username}_posts":"{username}"},# also invalidate "{username}_posts:{username}" cache)asyncdeferase_post(request:Request,username:str,id:int,current_user:Annotated[UserRead,Depends(get_current_user)],db:Annotated[AsyncSession,Depends(async_get_db)],):    ...# And now I'll also invalidate when I update the user:@router.patch("/{username}/post/{id}",response_model=PostRead)@cache("{username}_post_cache",resource_id_name="id",to_invalidate_extra={"{username}_posts":"{username}"})asyncdefpatch_post(request:Request,username:str,id:int,values:PostUpdate,current_user:Annotated[UserRead,Depends(get_current_user)],db:Annotated[AsyncSession,Depends(async_get_db)],):    ...

Warning

Note that addingto_invalidate_extra will not work forGET requests.

Invalidate Extra By Pattern

@router.get("/{username}/posts",response_model=PaginatedListResponse[PostRead])@cache(key_prefix="{username}_posts:page_{page}:items_per_page:{items_per_page}",resource_id_name="username",expiration=60,)asyncdefread_posts(request:Request,username:str,db:Annotated[AsyncSession,Depends(async_get_db)],page:int=1,items_per_page:int=10,):db_user=awaitcrud_users.get(db=db,schema_to_select=UserRead,username=username,is_deleted=False)ifnotdb_user:raiseHTTPException(status_code=404,detail="User not found")posts_data=awaitcrud_posts.get_multi(db=db,offset=compute_offset(page,items_per_page),limit=items_per_page,schema_to_select=PostRead,created_by_user_id=db_user["id"],is_deleted=False,    )returnpaginated_response(crud_data=posts_data,page=page,items_per_page=items_per_page)

@router.patch("/{username}/post/{id}")@cache("{username}_post_cache",resource_id_name="id",pattern_to_invalidate_extra=["{username}_posts:*"])asyncdefpatch_post(request:Request,username:str,id:int,values:PostUpdate,current_user:Annotated[UserRead,Depends(get_current_user)],db:Annotated[AsyncSession,Depends(async_get_db)],):    ...

Caution

Usingpattern_to_invalidate_extra can be resource-intensive on large datasets. Use it judiciously and consider the potential impact on Redis performance. Be cautious with patterns that could match a large number of keys, as deleting many keys simultaneously may impact the performance of the Redis server.

Client-side Caching

5.10 ARQ Job Queues

Tip

Very common candidates for background functions are calls to and from LLM endpoints (e.g. OpenAI or Openrouter). This is because they span tens of seconds and often need to be further parsed and saved.

Background task creation

asyncdefsample_background_task(ctx,name:str)->str:awaitasyncio.sleep(5)returnf"Task{name} is complete!"

from .functionsimportsample_background_taskfrom .your_moduleimportsample_complex_background_taskclassWorkerSettings:functions= [sample_background_task,sample_complex_background_task]    ...

Add the task to an endpoint

@router.post("/task",response_model=Job,status_code=201)asyncdefcreate_task(message:str):job=awaitqueue.pool.enqueue_job("sample_background_task",message)return {"id":job.job_id}@router.get("/task/{task_id}")asyncdefget_task(task_id:str):job=ArqJob(task_id,queue.pool)returnawaitjob.info()

Important

For any change to thesample_background_task to be reflected in the worker, you need to restart the worker (e.g. the docker container).

uv run arq src.app.core.worker.settings.WorkerSettings

Database session with background tasks

fromarq.workerimportWorkerfrom ...core.db.databaseimportasync_get_dbasyncdefstartup(ctx:Worker)->None:ctx["db"]=awaitanext(async_get_db())logging.info("Worker Started")asyncdefshutdown(ctx:Worker)->None:awaitctx["db"].close()logging.info("Worker end")

fromarq.workerimportWorkerasyncdefyour_background_function(ctx:Worker,post_id:int,    ...)->Any:db=ctx["db"]post=crud_posts.get(db=db,schema_to_select=PostRead,id=post_id)    ...

Warning

When using database sessions, you will want to use Pydantic objects. However, these objects don't mingle well with the seralization required by ARQ tasks and will be retrieved as a dictionary.

5.11 Rate Limiting

fromfastapiimportDependsfromapp.api.dependenciesimportrate_limiter_dependencyfromapp.core.utilsimportqueuefromapp.schemas.jobimportJob@router.post("/task",response_model=Job,status_code=201,dependencies=[Depends(rate_limiter_dependency)])asyncdefcreate_task(message:str):job=awaitqueue.pool.enqueue_job("sample_background_task",message)return {"id":job.job_id}

Warning

Do not forget to addapi/v1/... or any other prefix to the beggining of your path. For the structure of the boilerplate,api/v1/<rest_of_the_path>

{"data":[{"name":"free","id":1,"created_at":"2023-11-11T05:57:25.420360"},{"name":"pro","id":2,"created_at":"2023-11-12T00:40:00.759847"}],"total_count":2,"has_more":false,"page":1,"items_per_page":10}

{"data":[{"path":"api_v1_tasks_task","limit":10,"period":3600,"id":1,"tier_id":2,"name":"api_v1_tasks:10:3600"}],"total_count":1,"has_more":false,"page":1,"items_per_page":10}

curl -X POST'http://127.0.0.1:8000/api/v1/tasks/task?message=test' \-H'Authorization: Bearer <your-token-here>'

Tip

Since therate_limiter_dependency dependency uses theget_optional_user dependency instead ofget_current_user, it will not require authentication to be used, but will behave accordingly if the user is authenticated (and token is passed in header). If you want to ensure authentication, also useget_current_user if you need.

Warning

If a user does not have atier or the tier does not have a definedrate limit for the path and the token is still passed to the request, the defaultlimit andperiod will be used, this will be saved inapp/logs.

5.12 JWT Authentication

5.12.1 Details

# app/api/v1/login...response.set_cookie(key="refresh_token",value=refresh_token,httponly=True,# Prevent access through JavaScriptsecure=True,# Ensure cookie is sent over HTTPS onlysamesite="Lax",# Default to Lax for reasonable balance between security and usabilitymax_age=number_of_seconds,# Set a max age for the cookie)...

5.12.2 Usage

5.13 Admin Panel

5.13.1 Features

5.13.2 Access

http://localhost:8000/admin

5.13.3 Configuration

5.13.4 Customization

fromyour_app.modelsimportYourModelfromyour_app.schemasimportYourCreateSchema,YourUpdateSchemadefregister_admin_views(admin:CRUDAdmin)->None:# ... existing models ...admin.add_view(model=YourModel,create_schema=YourCreateSchema,update_schema=YourUpdateSchema,allowed_actions={"view","create","update","delete"}    )

# Handle models with problematic fields (e.g., TSVector)admin.add_view(model=Article,create_schema=ArticleCreate,update_schema=ArticleUpdate,select_schema=ArticleSelect,# Exclude problematic fields from read operationsallowed_actions={"view","create","update","delete"})# Password field handlingadmin.add_view(model=User,create_schema=UserCreateWithPassword,update_schema=UserUpdateWithPassword,password_transformer=password_transformer,# Handles password hashingallowed_actions={"view","create","update"})# Read-only modelsadmin.add_view(model=AuditLog,create_schema=AuditLogSchema,update_schema=AuditLogSchema,allowed_actions={"view"}# Only viewing allowed)

# Enable Redis sessions in your environmentCRUD_ADMIN_REDIS_ENABLED=trueCRUD_ADMIN_REDIS_HOST=localhostCRUD_ADMIN_REDIS_PORT=6379

5.14 Running

docker compose up

uv run uvicorn src.app.main:app --reload

uv run arq src.app.core.worker.settings.WorkerSettings

5.15 Create Application

# app/main.pyfrom .apiimportrouterfrom .core.configimportsettingsfrom .core.setupimportcreate_application# create_tables_on_start defaults to Trueapp=create_application(router=router,settings=settings,create_tables_on_start=False)

5.16 Opting Out of Services

# src/app/core/configimportosfromenumimportEnumfrompydantic_settingsimportBaseSettingsfromstarlette.configimportConfigcurrent_file_dir=os.path.dirname(os.path.realpath(__file__))env_path=os.path.join(current_file_dir,"..","..",".env")config=Config(env_path)...classSettings(AppSettings,PostgresSettings,CryptSettings,FirstUserSettings,TestSettings,RedisCacheSettings,ClientSideCacheSettings,RedisQueueSettings,RedisRateLimiterSettings,DefaultRateLimitSettings,CRUDAdminSettings,EnvironmentSettings,):passsettings=Settings()

classSettings(AppSettings,PostgresSettings,CryptSettings,FirstUserSettings,TestSettings,ClientSideCacheSettings,DefaultRateLimitSettings,EnvironmentSettings,):pass

version:'3.8'services:web:build:context:.dockerfile:Dockerfile# -------- replace with comment to run with gunicorn --------command:uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload# command: gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000env_file:      -./src/.env# -------- replace with comment if you are using nginx --------ports:      -"8000:8000"# expose:#   - "8000"depends_on:      -db      -redisvolumes:      -./src/app:/code/app      -./src/.env:/code/.envdb:image:postgres:13env_file:      -./src/.envvolumes:      -postgres-data:/var/lib/postgresql/data# -------- replace with comment to run migrations with docker --------expose:      -"5432"# ports:#  - 5432:5432volumes:postgres-data:redis-data:#pgadmin-data:

6. Running in Production

6.1 Uvicorn Workers with Gunicorn

command: gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000

# docker-compose.yml# -------- replace with comment to run with gunicorn --------command:uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload# command: gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000

# docker-compose.yml# -------- replace with comment to run with uvicorn --------# command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reloadcommand:gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000

# DockerfileCMD ["uvicorn","app.main:app","--host","0.0.0.0","--port","8000","--reload"]# CMD ["gunicorn", "app.main:app", "-w", "4", "-k", "uvicorn.workers.UvicornWorker". "-b", "0.0.0.0:8000"]

# Dockerfile# CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--reload"]CMD ["gunicorn","app.main:app","-w","4","-k","uvicorn.workers.UvicornWorker"."-b","0.0.0.0:8000"]

Caution

Do not forget to set theENVIRONMENT in.env toproduction unless you want the API docs to be public.

6.2 Running with NGINX

# docker-compose.yml...# -------- uncomment to run with nginx --------# nginx:#   image: nginx:latest#   ports:#     - "80:80"#   volumes:#     - ./default.conf:/etc/nginx/conf.d/default.conf#   depends_on:#     - web...

# docker-compose.yml...#-------- uncomment to run with nginx --------nginx:image:nginx:latestports:      -"80:80"volumes:      -./default.conf:/etc/nginx/conf.d/default.confdepends_on:      -web...

# docker-compose.ymlservices:web:...# -------- Both of the following should be commented to run with nginx --------command:uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload# command: gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000

# docker-compose.ymlservices:web:...# -------- Both of the following should be commented to run with nginx --------# command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload# command: gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000

# DockerfileCMD ["uvicorn","app.main:app","--host","0.0.0.0","--port","8000","--reload"]# CMD ["gunicorn", "app.main:app", "-w", "4", "-k", "uvicorn.workers.UvicornWorker". "-b", "0.0.0.0:8000"]

6.2.1 One Server

# default.conf# ---------------- Running With One Server ----------------server {    listen 80;    location / {        proxy_pass http://web:8000;        proxy_set_header Host $host;        proxy_set_header X-Real-IP $remote_addr;        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;        proxy_set_header X-Forwarded-Proto $scheme;    }}

6.2.2 Multiple Servers

# default.conf# ---------------- Running With One Server ----------------...# ---------------- To Run with Multiple Servers, Uncomment below ----------------upstream fastapi_app {    server fastapi1:8000;  # Replace with actual server names or IP addresses    server fastapi2:8000;    # Add more servers as needed}server {    listen 80;    location / {        proxy_pass http://fastapi_app;        proxy_set_header Host $host;        proxy_set_header X-Real-IP $remote_addr;        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;        proxy_set_header X-Forwarded-Proto $scheme;    }}

Warning

Note that we are usingfastapi1:8000 andfastapi2:8000 as examples, you should replace it with the actual name of your service and the port it's running on.

7. Testing

7.1 Writing Tests

touch tests/test_items.py

importpytestfromunittest.mockimportAsyncMock,patchfromsrc.app.api.v1.usersimportwrite_userclassTestWriteUser:@pytest.mark.asyncioasyncdeftest_create_user_success(self,mock_db,sample_user_data):"""Test successful user creation."""withpatch("src.app.api.v1.users.crud_users")asmock_crud:mock_crud.exists=AsyncMock(return_value=False)mock_crud.create=AsyncMock(return_value=Mock(id=1))result=awaitwrite_user(Mock(),sample_user_data,mock_db)assertresult.id==1mock_crud.create.assert_called_once()

7.2 Running Tests

uv run pytest

uv run pytest tests/test_user_unit.py

uv run pytest tests/test_user_unit.py

uv run pytest -v

uv run pytest tests/test_user_unit.py::TestWriteUser::test_create_user_success

7.3 Test Configuration

[tool.pytest.ini_options]filterwarnings = ["ignore::PendingDeprecationWarning:starlette.formparsers",]

7.4 Test Structure

7.5 Benefits of Our Approach

8. Contributing

9. References

10. License

11. Contact