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
fromsqlalchemyimportColumn, Integer, String, create_enginefromsqlalchemy.ormimportDeclarativeMeta, declarative_base, sessionmakerfromfastapi_sqlalchemyimportSQLAlchemydb=SQLAlchemy(url="sqlite:///example.db")
#Define User classclassUser(db.Base):
__tablename__="items"id=Column(Integer, primary_key=True)
name=Column(String)
email=Column(String)
def__repr__(self):
returnf"User(id={self.id}, name='{self.name}',email='{self.email}')"fromfastapiimportFastAPIfrommodelsimportUser, dbfrompydanticimportBaseModelfromfastapi_sqlalchemyimportDBSessionMiddlewareapp=FastAPI()
# Add SQLAlchemy session middleware to manage database sessionsapp.add_middleware(DBSessionMiddleware, db=db)
# Endpoint to retrieve all users@app.get("/users")defget_users():
""" Retrieve a list of all users. Returns: List[User]: A list of User objects. """returnUser.query.all()
# Pydantic model for creating new usersclassUserCreate(BaseModel):
name: stremail: str# Endpoint to add a new user@app.post("/add_user")defadd_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:
importpytzfromapscheduler.schedulers.asyncioimportAsyncIOScheduler# other schedulers are availablefromfastapiimportFastAPIfrommodelsimportUser, dbfromfastapi_sqlalchemyimportDBSessionMiddlewareapp=FastAPI()
app.add_middleware(DBSessionMiddleware, db_url="sqlite:///example.db")
@app.on_event('startup')asyncdefstartup_event():
scheduler=AsyncIOScheduler(timezone=pytz.utc)
scheduler.start()
scheduler.add_job(count_users_task, "cron", hour=0) # runs every night at midnightdefcount_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:withdb():
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 endedreturnusersYou can define custom BaseModels, or extend the built in ModelBase to provide extended shared functionality for you database models.
importinspectfromtypingimportListfromsqlalchemyimportColumnfromfastapi_sqlalchemyimportModelBaseclassBaseModel(ModelBase):
@classmethoddefnew(cls, **kwargs):
obj=cls(**kwargs)
obj.save()
returnobj@classmethoddefget(cls, **kwargs):
result: cls=cls.query.filter_by(**kwargs).first()
returnresult@classmethoddefget_all(cls, **kwargs):
result: List[cls] =cls.query.filter_by(**kwargs).all()
returnresultdefupdate(self, **kwargs):
forcolumn, valueinkwargs.items():
setattr(self, column, value)
self.save()
returnselfAs 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
fromsqlalchemyimportColumn, Integer, String, create_enginefromsqlalchemy.ormimportdeclarative_base, sessionmakerfromfastapi_sqlalchemyimportModelBase, SQLAlchemydb=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 propertiesclassUser(ModelBase, db.Base):
__tablename__="users"id=Column(Integer, primary_key=True)
name=Column(String)
email=Column(String)
def__repr__(self):
returnf"User(id={self.id}, name='{self.name}',email='{self.email}')"fromfastapiimportFastAPIfromfastapi_sqlalchemyimportDBSessionMiddleware# middleware helperfromfastapi_sqlalchemyimportdb# an object to provide global access to a database sessionfromapp.modelsimportUserapp=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")defget_users():
users=db.session.query(User).all()
returnusersNote 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:
importpytzfromapscheduler.schedulers.asyncioimportAsyncIOScheduler# other schedulers are availablefromfastapiimportFastAPIfromfastapi_sqlalchemyimportdbfromapp.modelsimportUser, UserCountapp=FastAPI()
app.add_middleware(DBSessionMiddleware, db_url="sqlite:///example.db")
@app.on_event('startup')asyncdefstartup_event():
scheduler=AsyncIOScheduler(timezone=pytz.utc)
scheduler.start()
scheduler.add_job(count_users_task, "cron", hour=0) # runs every night at midnightdefcount_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:withdb():
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 endedreturnusers