Made withFastAPITutorial

How to Use Async SQLAlchemy in FastAPI

P

Pretty Printed

Creator on the FastAPI channel

Watch on YouTube ↗

42K viewsMay 27, 202315:20 runtime

Learn how to integrate async SQLAlchemy into your FastAPI applications by converting a synchronous project to an asynchronous one. This guide demonstrates the necessary changes, from installing the required AIO SQLite driver to updating your database connection URI. Discover how…

Chapters

#fastapi#tutorial#backend

Using Async SQLAlchemy in FastAPI

This document outlines the key steps to convert a synchronous FastAPI application using SQLAlchemy to a fully asynchronous one.

## Initial Setup & Dependencies

Install Async Driver: An async database driver is required. For SQLite, you can use aiosqlite.
```
pip install aiosqlite
```
Update Database URL: Modify the database connection string to specify the async driver.
- Sync: sqlite:///example.db
- Async: sqlite+aiosqlite:///example.db

## Core Async Components

Engine: Replace create_engine with create_async_engine from sqlalchemy.ext.asyncio.
Session Maker: Replace sessionmaker with async_session_maker.
Type Hinting: Use AsyncSession for type hints in your dependencies and endpoints.

## Database Schema Creation

Important
Top-level synchronous calls like Base.metadata.create_all(engine) will not work in an async application. The schema must be created within an async context.

A common pattern is to create the tables on the first API call inside the get_db dependency.

# In your async get_db dependency
async with engine.begin() as conn:
    await conn.run_sync(Base.metadata.create_all)

## Updating FastAPI Endpoints

Async Functions: All endpoints that perform database I/O must be defined with async def.
Await Operations: Any database call that interacts with the database must be awaited.
- await db.commit()
- await db.refresh(my_object)
- await db.execute(...)

## Modern SQLAlchemy 2.0 Syntax

Async support is built on the SQLAlchemy 2.0 paradigm, which introduces a new way to define models and write queries.

Model Definition

Models should use type-annotated columns with Mapped and mapped_column from sqlalchemy.orm.

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = 'users'
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(unique=True)

Querying

The legacy db.query() method is replaced by a two-step process using select().

Build the Query: Create a query object using select() from sqlalchemy.
Execute and Fetch: Use db.execute() to run the query and then process the results.

# Get all users
from sqlalchemy import select

result = await db.execute(select(User))
users = result.scalars().all()

Note
The .scalars() method is used to get the primary ORM objects from the result, discarding the tuple structure that would otherwise be returned. .all() fetches all results into a list.

Introduction: Sync to Async SQLAlchemy

Hey everyone. In today's video, I'm going to show you how to use async SQLAlchemy in your FastAPI projects. So to do this, I'm going to convert this example that I've already written here which is using regular SQLAlchemy and I'm going to convert it over to async SQLAlchemy so you can see where things need to change.

And before I get into that, just know that if you need help with anything like this, so using async SQLAlchemy in your FastAPI project or anything else with your FastAPI projects, I do have a coaching program where I work with people one-on-one. So you can learn more about that at prettyprinted.com/coaching or you can go to the link in the description below.

Reviewing the Synchronous App Code

So to start, what I want to do is I want to tell you what I have in this example. So this is the regular SQLAlchemy and then I'll convert it. So after the imports, what I'm doing is I'm creating the engine for SQLAlchemy. So this will allow me to eventually create sessions. So here I have the session maker using the engine, and I'm connecting to a SQLite database which hasn't been created yet. That will be created in just a moment. I need to check same thread as false because of the way that FastAPI works internally. This is only necessary for SQLite databases. So if you use like Postgres, you don't need this particular argument.

So I have this session maker function, which will create this session local class which I can use to instantiate a DB object that will then interact with a database and you'll see that below in just a second. I have this Base class here which is from declarative base. This will allow me to convert regular classes into SQLAlchemy models, which I have here for the class User. So I'm creating a table and a model called User. The table name will be users here, and then I have an ID and a username column. Then I'm creating everything. So this will create the database, will actually just starting this file without this will create the database, but this will create all the models in the database. So just this one user model.

Explaining Endpoints and Dependencies

Then I have this helper function called get_db. So it uses the SessionLocal class that I created from the session maker. When I instantiate this, I get a DB object back that I can use to interact with the database. What I'm going to do is I'm going to yield it here so I can use it, and then once I finish using it, it will return back to this function and then it will close the database so I don't have any problems with my connections. I have this UserBase for Pydantic just to create new users in the database.

And then I have two endpoints. So one endpoint is for creating a user, so /user, post. It's going to take in the user base which has a username, and then it's going to use Depends to automatically load in the DB object by calling get_db every time this endpoint runs. So with the DB object here, I can then instantiate a user and then add it to the session using db.add. So that user I can call commit and then I can refresh to put the extra information on the user object here, which will be the ID in this case because it's generated when I save to the database, and I'll return this back to the caller. So we'll see that in just a second.

And then this one is similar in the sense that it has the Depends here, and I'm calling get_db which will return my DB object. I'm going to use that to query the database for all the users and then I'm just going to return them as a list and it will automatically be serialized to JSON even though this will be a list of SQLAlchemy objects.

Demonstrating the Synchronous App

So now let me get this started. I will use uvicorn, and then the name of the file is app, and then the name of the object is app, and I'll just put reload on. And I'll go to localhost:8000 on my browser here. /docs. Okay, so we see Swagger here. And I have the two endpoints, so one for creating a user and one for getting users. So if I call the get for get all the users, I see I get nothing, users is blank here. Let me just increase the size there. So we see I get nothing for users. But if I go to this one, the other user, and try it out, I can put in a username, so I'll use my name, Anthony, hit execute, and we see it returns username Anthony and id: 1. So that ID 1 came from the database. Then I can go to get users again, I can execute, and now I see my user Anthony is one of the users that returns. So this is using regular SQLAlchemy, and it's also using just regular functions. These aren't async functions for the endpoints that I'm defining. So now that I have all that, so that is the synchronous example, let's get into the async example.

Installing the Async Driver and Updating the URI

So the first thing I want to do is I want to install something to make this work. So I need a driver that can interact with a SQLite database in an asynchronous manner. So for that, I'm going to use something called aiosqlite. There are other ones, but this is the one I'm going to use. So pip install aiosqlite. And I'll just install that. And now that I have that, I want to change the URI for my database. So instead of just sqlite, I'm going to do plus and then aio sqlite, just like that. So this will tell SQLAlchemy that I want to use this particular driver to interact with my database.

Setting Up the Async Engine and Session

So now that I have that, let me bring in the SQLAlchemy things that I need. So first I need the stuff from the asyncio section of SQLAlchemy. So from sqlalchemy.ext.asyncio, I want to import a function called create_async_engine. And I also want to import a function called async_session_maker. So what I can do is I can take this function create_async_engine and I can just replace the create_engine that I have here. So now this is create_async_engine and this engine object will now be for an async engine. And for the session_maker, I want to replace that with the async_session_maker, so I'll just put that there. So now the SessionLocal will be for async sessions, not regular sessions.

For the declarative base, this can remain the same. I am going to change this a little bit later, but for now this can remain the same. And also the types of columns can remain the same. I will change this later to give you an example at the end of the video, but for now these two are fine. So this is the first thing that needs to change. So in the sync example, I could just put this base.metadata.create_all here because it's just executing the code from top to bottom. So when it gets to this line, it runs it. But when I'm using async, I can't simply await something here because I need to be inside of an event loop. So I can't just put an await something here. So what I'm going to do is I'm going to get rid of this, how about I just completely get rid of it, and I'm going to put the create_all inside of the get_db. So what I can do here is first I can change this get_db to async get_db, and what I want to do is I want to say async with and then engine.begin() as connection and then in this block I can then await connection.run_sync(base.metadata.create_all). So this will call this create_all using run_sync. So because I'm using async sql alchemy, I then need to run this synchronously for this particular thing, creating all the things in the database, but I still have to use the async with block for this just because I'm using the async engine here. So in the get_db, the SessionLocal part is still the same, and for closing the database, I want to await closing it because I can't close it just by calling db.close anymore.

Okay, so now the user base is going to remain the same, this is just pydantic.

Converting Endpoints to Async

And now let's go down to the two endpoints that I have. So the first thing I want to do is I want to change both of them to async. So async def, just like that. And for creating a user, the part that I have to await is committing. So here after creating a user, everything here happens outside of the database but as soon as I go to commit something, I have to await. So await db.commit() and I also have to await db.refresh.

And down here, this is going to be a little bit different, so I'm not going to touch this one for now. I'll just create a new user and show you how it works. So let me go ahead and start this. Oh, and by the way, the Session here is just for type checking. So it's not actually used. So what I can do is I can go up to the top here and I can import async session. And I can use that async session for type checking, so I can replace the session here and or in the session there. Okay, so let's start this. And everything starts up properly. So now let's go to the endpoint for creating a user. So let me just refresh, so everything reloads. I'll create another user and the username will be 'pretty printed' here. So I'll execute and we see I get id: 2 and username 'pretty printed'. So this is using the async version. So everything here works fine.

And just in case you don't believe me, what I'll do is I'll take away the awaits and then I'll try running this again. So I'll create a third user called 'python_user'. Hit execute, and now I get an error and it says that the error here is that the user isn't persistent in the session. But more importantly the warning that it's going to give me if I go above, it should be something like I'm trying to call it without awaiting it. Yeah, so async_session.commit was never awaited. So I have to await to actually save the user to the database. So let me go ahead and await that again and go back here and create the user. And now we see ID3 and python_user has been added to the database, and there are no errors.

Refactoring Models for SQLAlchemy 2.0

So for this one, I need to change the way this query works because in SQLAlchemy 2.0 which supports async SQLAlchemy, it also changes the way that queries are run and it also changes the way that the standard way of defining classes are done. So what I'm going to do is I'm going to make those modifications now so I can run this query down at the bottom. So the first thing is the base here. So instead of calling a base just like this, what I can do is I can import a class directly. So DeclarativeBase, just like that. And I'm going to inherit from this DeclarativeBase or actually not there, let's just say here. So I'll call this class Base and I inherit it from DeclarativeBase and I'll just pass, right? So this part works the same, but I no longer have to call this function declarative_base, I can just remove it here.

The next thing I want to do is I want to change the way that I define the columns. So before you would import like Column and Integer and so on to define columns. But now what you do is you use type checking. So I want to define a column called ID, and I need to give it a type. So it's going to be Mapped, and I need to import this Mapped from sqlalchemy.orm. So mapped just like that. So it's Mapped, and then the type inside of Mapped is the type that I want the column to be. So in this case I want it to be an int. And then this will be equal to any additional information that I want to pass. So for this, I need a function called mapped_column and I'll just put this here. And inside of mapped_column, then I can say that the primary_key is True, and that's it for this one. So it just has the primary key. And then I can get rid of the original ID.

So now let me add a username. So username is similar. So username, and then once again I need Mapped, and then the type, so this will be a str, because it's a string, and then I'll use mapped_column again to say that unique is True. If I didn't have anything other than just the string here, then I wouldn't need the mapped_column, but because I want to say that this should be unique, I can use the mapped_column function here to say unique is True. So I'll remove that.

Implementing the Modern Async Query Style

And now at the bottom, what I want to do is I want to change this query. So instead of running the query like this to get the users, the first thing I need to do is I need to tell the database I want to execute a query. So this is going to be kind of similar to if you were using like the driver directly, where you first execute on a cursor and then you get the results from the cursor. So what I want to do is I'll say like results = and then I need to await db.execute(). And inside of here, I want to use a function called select() and I'll pass in the User class. So I need to import this select from SQL Alchemy. So just up here, select lowercase.

And now that I have the results, I want to convert them to a format that I can use. So to get them in the same format as users before, I can do results.scalars().all(). So this converts the results to something that doesn't have the extra SQLAlchemy metadata and it gives me all the results. So it's like a two-step process. So this will just replace that. So what I'll do is I'll recreate the database so we can see everything working. So I'll reload the app and now I'll refresh this page. We see I don't have a database yet because the only time this gets run is when I call get_db, and the only time get_db gets called is when I use one of the endpoints. So what I'll do is I'll query for all the users. Try it out, execute. It returns none. And we see the database is here now. And now let me go ahead and add a user. So down here, add a user. Username is going to be Anthony. And I'll execute here. We see it created username Anthony with id 1. Now if I go back here, hit execute, I see Anthony is in the list of users. And let me go ahead and create one more user, 'pretty printed'. Hit execute, it was created properly. I'll hit execute up here and now we see Anthony and pretty printed users, ID one and two.

Final Thoughts on the New Syntax

So that's all that you need to do to get async SQLAlchemy working in your FastAPI project. So of course you need async functions and of course you need to await the things that need to be awaited for SQLAlchemy. And you'll need to convert to using the new style of SQLAlchemy. And I personally don't like this new approach to writing queries in SQL Alchemy, even though I understand how it's closer to the model of interacting with databases directly in terms of writing queries, but I just felt like the old approach was so much more convenient, you know, it actually made writing queries a little bit easier in a lot of cases. And then you always had the fallback method of writing more verbose queries if you wanted to get something really complicated done. But for simple cases, I really like the old style of writing queries. So let me know what you think of the new style of SQL Alchemy and let me know if you have any questions about using async SQLAlchemy and FastAPI. So that's it for this video. If you like this video, please give me a thumbs up. And if you have subscribed to my channel already, please subscribe. So thank you for watching and I will talk to you next time.