FastAPI-SQLAlchemy provides a simple integration between FastAPI and SQLAlchemy in your application. It gives access to useful helpers to facilitate the completion of common tasks.
Install and update using pip:
$ pip install fastapi-sqlalchemy
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import DeclarativeMeta, declarative_base, sessionmaker
from fastapi_sqlalchemy import SQLAlchemy
db = SQLAlchemy(url="sqlite:///example.db")
#Define User class
class User(db.Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True)
name = Column(String)
email = Column(String)
def __repr__(self):
return f"User(id={self.id}, name='{self.name}',email='{self.email}')"from fastapi import FastAPI
from models import User, db
from pydantic import BaseModel
from fastapi_sqlalchemy import DBSessionMiddleware
app = FastAPI()
# Add SQLAlchemy session middleware to manage database sessions
app.add_middleware(DBSessionMiddleware, db=db)
# Endpoint to retrieve all users
@app.get("/users")
def get_users():
"""
Retrieve a list of all users.
Returns:
List[User]: A list of User objects.
"""
return User.query.all()
# Pydantic model for creating new users
class UserCreate(BaseModel):
name: str
email: str
# Endpoint to add a new user
@app.post("/add_user")
def add_user(user_data: UserCreate):
"""
Add a new user to the database.
Args:
user_data (UserCreate): User data including name and email.
Returns:
dict: A message indicating the success of the operation.
"""
user = User(**user_data.model_dump())
print(user)
user.save()
return {"message": "User created successfully"}You can initialize the SQLAlchemy() class similar to the way flask-sqlalchemy, this allows for multiple database connections to work at the same time.
Sometimes it is useful to be able to access the database outside the context of a request, such as in scheduled tasks which run in the background:
import pytz
from apscheduler.schedulers.asyncio import AsyncIOScheduler # other schedulers are available
from fastapi import FastAPI
from models import User, db
from fastapi_sqlalchemy import DBSessionMiddleware
app = FastAPI()
app.add_middleware(DBSessionMiddleware, db_url="sqlite:///example.db")
@app.on_event('startup')
async def startup_event():
scheduler = AsyncIOScheduler(timezone=pytz.utc)
scheduler.start()
scheduler.add_job(count_users_task, "cron", hour=0) # runs every night at midnight
def count_users_task():
"""Count the number of users in the database and save it into the user_counts table."""
# we are outside of a request context, therefore we cannot rely on ``DBSessionMiddleware``
# to create a database session for us. Instead, we can use the same ``db`` object and
# use it as a context manager, like so:
with db():
user_count = User.query.count()
user_count = UserCount(user_count)
user_count.save()
# no longer able to access a database session once the db() context manager has ended
return usersYou can define custom BaseModels, or extend the built in ModelBase to provide extended shared functionality for you database models.
import inspect
from typing import List
from sqlalchemy import Column
from fastapi_sqlalchemy import ModelBase
class BaseModel(ModelBase):
@classmethod
def new(cls, **kwargs):
obj = cls(**kwargs)
obj.save()
return obj
@classmethod
def get(cls, **kwargs):
result: cls = cls.query.filter_by(**kwargs).first()
return result
@classmethod
def get_all(cls, **kwargs):
result: List[cls] = cls.query.filter_by(**kwargs).all()
return result
def update(self, **kwargs):
for column, value in kwargs.items():
setattr(self, column, value)
self.save()
return selfAs you can see the above BaseModel class adds support for various common functions and operations.
Note the only change that you need to make is to add the db.Base inheritance to each of your model classes
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import declarative_base, sessionmaker
from fastapi_sqlalchemy import ModelBase, SQLAlchemy
db = SQLAlchemy(url="sqlite:///example.db")
# Define the User class representing the "users" database table
# Using the SQLAlchemy Base property instead of defining your own
# And inheriting from the BaseModel class for type hinting and helpful builtin methods and properties
class User(ModelBase, db.Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String)
email = Column(String)
def __repr__(self):
return f"User(id={self.id}, name='{self.name}',email='{self.email}')"from fastapi import FastAPI
from fastapi_sqlalchemy import DBSessionMiddleware # middleware helper
from fastapi_sqlalchemy import db # an object to provide global access to a database session
from app.models import User
app = FastAPI()
app.add_middleware(DBSessionMiddleware, db_url="sqlite:///example.db")
# once the middleware is applied, any route can then access the database session
# from the global ``db``
@app.get("/users")
def get_users():
users = db.session.query(User).all()
return usersNote that the session object provided by db.session is based on the
Python3.7+ ContextVar. This means that each session is linked to the
individual request context in which it was created.
Sometimes it is useful to be able to access the database outside the context of a request, such as in scheduled tasks which run in the background:
import pytz
from apscheduler.schedulers.asyncio import AsyncIOScheduler # other schedulers are available
from fastapi import FastAPI
from fastapi_sqlalchemy import db
from app.models import User, UserCount
app = FastAPI()
app.add_middleware(DBSessionMiddleware, db_url="sqlite:///example.db")
@app.on_event('startup')
async def startup_event():
scheduler = AsyncIOScheduler(timezone=pytz.utc)
scheduler.start()
scheduler.add_job(count_users_task, "cron", hour=0) # runs every night at midnight
def count_users_task():
"""Count the number of users in the database and save it into the user_counts table."""
# we are outside of a request context, therefore we cannot rely on ``DBSessionMiddleware``
# to create a database session for us. Instead, we can use the same ``db`` object and
# use it as a context manager, like so:
with db():
user_count = db.session.query(User).count()
db.session.add(UserCount(user_count))
db.session.commit()
# no longer able to access a database session once the db() context manager has ended
return users