FastAPI + SQLAlchemy Tutorial

by

The code for this article can be found on my GitHub: https://github.com/jmgraff/fastapi-sqlalchemy-sqlite-example

FastAPI has made creating an API for your app incredibly simple. SQLAlchemy has also done the same for interacting with databases. Using both of them together makes it very simple to have your API store and retrieve data from a number of different database technologies. If you’re new to SQLAlchemy or want to see how to use it with a more full-featured database, take a look at our SQLAlchemy + PostgreSQL tutorial before continuing on.

In this article, we’re going to create a FastAPI app that interacts with a SQLite database using SQLAlchemy. Why SQLite? For one, it’s a simple file-based database which makes this tutorial easier to follow. It’s also surprisingly good for low to moderate traffic web apps, which this will absolutely be.

Let’s get started by taking a look at what Python packages we need.

Requirements

Here’s the requirements.txt file from the project:

aiosqlite
fastapi
sqlalchemy
uvicorn[standard]
container/requirements.txt

The packages we need are:

  • aiosqlite – This is the asyncio SQLite driver that we’ll be using. FastAPI uses asyncio to efficiently handle multiple web requests at once, so we need an asyncio-compatible version of sqlite to take advantage of that.
  • fastapi – Self explanatory
  • sqlalchemy – This will be our ORM library we use to interact with the SQLite database
  • uvicorn[standard] – This is our web server

Docker Compose File

When developing web apps, I greatly prefer to use Docker compose to manage the stack. It makes setting up and tearing down the development environment very simple.

Here’s the compose file I’ll be using:

services:
  api:
    image: api
    build: container
    volumes:
      - ./container/src:/opt/container
      - ./data:/opt/data
    ports:
      - 8080:8080
docker-compose.yaml

We have just one service, the API. We map a couple filesystem volumes in there for development purposes. In production, we’d probably use a named volume. We also expose port 8080 so we can interact with our API.

Dockerfile

We are building a custom container that has our code and all requirements installed. Here’s the Dockerfile for that:

from python

copy ./requirements.txt .
run pip install -r requirements.txt && rm requirements.txt

copy ./src /opt/container

workdir /opt/container

entrypoint ["python", "main.py"]
container/Dockerfile

We start with the python image as a base, since it has the latest and greatest Python already installed, as well as pip, which we’ll need. We then copy our requirements.txt file into the container, install the packages with it, then delete it (we won’t be needing it anymore after this).

We then copy our source code into /opt/container. This location doesn’t mater much, but it has to be consistent with what’s mapped in your docker-compose file. /opt/container is just a pattern I’ve used in my apps for a while and it’s how I prefer to do things.

We change our workdir to where we put our source code, and then tell Docker that the entry point of our app should be running the main.py file with the python interpreter.

The Full App Code

Before we dig into each individual part of this app, I always find it useful when tutorials just show me the entire code. Sometimes all you need is a reference.

So, here it is:

import typing
import pathlib

import uvicorn

from fastapi import FastAPI, Depends

from sqlalchemy import Column, Integer, String, select
from sqlalchemy.orm import declarative_base
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from pydantic import BaseModel

Base = declarative_base()

# This is our SQLAlchemy model used with the database
class Thing(Base):
    __tablename__ = "thing"
    id = Column(Integer, primary_key=True)
    text = Column(String, nullable=False)

pathlib.Path("/opt/data").mkdir(exist_ok=True)
engine = create_async_engine("sqlite+aiosqlite:////opt/data/db.sqlite3")
async_session_maker = async_sessionmaker(engine)

async def create_db_and_tables():
    async with engine.begin() as conn:
        # We drop all tables on startup just for demonstration purposes.
        # Don't do this in production!
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)

async def get_async_session() -> typing.AsyncGenerator[AsyncSession, None]:
    async with async_session_maker() as session:
        yield session


app = FastAPI()

# This is our Pydantic model used with the API
class ThingModel(BaseModel):
    id: typing.Optional[int]
    text: str

# We use the "startup" event to create our database and tables
@app.on_event("startup")
async def on_startup(session: AsyncSession = Depends(get_async_session)):
    await create_db_and_tables()

@app.get("/things")
async def get_things(session: AsyncSession = Depends(get_async_session)) -> typing.List[ThingModel]:
    things = await session.execute(select(Thing))
    return [ThingModel(id=thing.id, text=thing.text) for thing in things.scalars()]

@app.get("/things/{id}")
async def get_thing(id: str, session: AsyncSession = Depends(get_async_session)):
    thing = await session.get(Thing, id)
    if thing:
        return ThingModel(id=thing.id, text=thing.text)

@app.post("/things")
async def post_thing(thing: ThingModel, session: AsyncSession = Depends(get_async_session)):
    session.add(Thing(text=thing.text))
    await session.commit()

@app.delete("/things/{id}")
async def delete_thing(id: str, session: AsyncSession = Depends(get_async_session)):
    thing = await session.get(Thing, id)
    await session.delete(thing)
    await session.commit()

@app.put("/things/{id}")
async def put_thing(id: str, new_thing: ThingModel, session: AsyncSession = Depends(get_async_session)):
    thing = await session.get(Thing, id)
    thing.text = new_thing.text
    await session.commit()

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8080, reload=True)
container/src/main.py

I put all of it in the main.py file for simplicity’s sake. If you are making an app that will see real world use, I’d definitely split this up. It’ll be more maintainable for you and your team that way.

Create the Schema

Let’s get started by creating our database schema:

from sqlalchemy.orm import declarative_base
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from pydantic import BaseModel

Base = declarative_base()

# This is our SQLAlchemy model used with the database
class Thing(Base):
    __tablename__ = "thing"
    id = Column(Integer, primary_key=True)
    text = Column(String, nullable=False)

pathlib.Path("/opt/data").mkdir(exist_ok=True)
engine = create_async_engine("sqlite+aiosqlite:////opt/data/db.sqlite3")
async_session_maker = async_sessionmaker(engine)
container/src/main.py

SQLAlchemy makes this very easy. We’re going to be using the “declarative” style of ORM in the library, so we start by creating a Base class that our table classes will inherit from.

Then, we define our one and only table. A real app will probably have a lot of these, with complex relationships. For us, we have one table called “thing”, represented by the class Thing you see above. It has a couple columns: id which is an integer and primary_key, and the text column, which is just a string. Note that nullable=False means we must always specify a value for this field every time we create a new Thing.

Set up Your Database Connection

Now that we have a schema to use, we have to connect to our database:

    id = Column(Integer, primary_key=True)
    text = Column(String, nullable=False)

pathlib.Path("/opt/data").mkdir(exist_ok=True)
engine = create_async_engine("sqlite+aiosqlite:////opt/data/db.sqlite3")
async_session_maker = async_sessionmaker(engine)

async def create_db_and_tables():
    async with engine.begin() as conn:
container/src/main.py

Since SQLite is a file based database, we’re not really “connecting”, but just opening the file. That’s transparent to us though. We open it with a URI the same way we would when connecting to a full fledged database server. We are using create_async_engine here rather than create_engine you might see in other tutorials – this is because we want to make use of asyncio‘s features when talking to the databse.

We have to create an async_session_maker function with the factory method async_sessionmaker. This looks a little confusing because the names are basically the same, just without a space in one of them. Don’t get tripped up by this. You can name the function returned by async_sessionmaker(engine) anything you want – we just named it async_session_maker here because, well, that’s what it is, and it’s a descriptive name.

Build the Tables

We now have to define a function to build our tables:

pathlib.Path("/opt/data").mkdir(exist_ok=True)
engine = create_async_engine("sqlite+aiosqlite:////opt/data/db.sqlite3")
async_session_maker = async_sessionmaker(engine)

async def create_db_and_tables():
    async with engine.begin() as conn:
        # We drop all tables on startup just for demonstration purposes.
        # Don't do this in production!
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)

async def get_async_session() -> typing.AsyncGenerator[AsyncSession, None]:
    async with async_session_maker() as session:
        yield session
container/src/main.py

This is pretty straightforward. We’re connecting to the database in order to drop all existing tables (this effectively wipes the database) and then create all the tables associated with our schema. The Base base class knows about all the tables we created that inherit from it.

Note that you would not drop all tables in production. This is only being done here to make sure we start from a known state every time we launch the app in development.

Get an AsyncSession

We now have to define a function that will get us an asynchronous session. This function will be called to provide the session to our app’s routes, which we’ll define later:

async def create_db_and_tables():
    async with engine.begin() as conn:
        # We drop all tables on startup just for demonstration purposes.
        # Don't do this in production!
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)

async def get_async_session() -> typing.AsyncGenerator[AsyncSession, None]:
    async with async_session_maker() as session:
        yield session


app = FastAPI()
container/src/main.py

Create your FastAPI App

Now, the easiest line in the app:

    async with async_session_maker() as session:
        yield session


app = FastAPI()

# This is our Pydantic model used with the API
class ThingModel(BaseModel):
    id: typing.Optional[int]
    text: str
container/src/main.py

That’s all you have to do to create a FastAPI app. We’ll use the app object to define our routes later.

Create your Pydantic Model

FastAPI uses Pydantic models to receive and send data to clients via the API. So, even though we already created a Thing object for SQLAlchemy, we’ll also have to create one for FastAPI to use:


app = FastAPI()

# This is our Pydantic model used with the API
class ThingModel(BaseModel):
    id: typing.Optional[int]
    text: str

# We use the "startup" event to create our database and tables
@app.on_event("startup")
async def on_startup(session: AsyncSession = Depends(get_async_session)):
    await create_db_and_tables()
container/src/main.py

We say id is optional here because it won’t always have anything in it, such as when creating a new Thing. This will make more sense when we define our routes.

Build your Schema on Startup

The function we defined to build our schema has to be called when the app starts up:


# This is our Pydantic model used with the API
class ThingModel(BaseModel):
    id: typing.Optional[int]
    text: str

# We use the "startup" event to create our database and tables
@app.on_event("startup")
async def on_startup(session: AsyncSession = Depends(get_async_session)):
    await create_db_and_tables()

@app.get("/things")
async def get_things(session: AsyncSession = Depends(get_async_session)) ->
container/src/main.py

We do this by hooking into the “startup” even with FastAPI’s on_event API.

Create your CRUD Routes

Now, it’s time to create the routes that our clients will use to interact with our API.

GET All

The first route we’re going to define is how to get all rows from the things table:

# We use the "startup" event to create our database and tables
@app.on_event("startup")
async def on_startup(session: AsyncSession = Depends(get_async_session)):
    await create_db_and_tables()

@app.get("/things")
async def get_things(session: AsyncSession = Depends(get_async_session)) -> typing.List[ThingModel]:
    things = await session.execute(select(Thing))
    return [ThingModel(id=thing.id, text=thing.text) for thing in things.scalars()]

@app.get("/things/{id}")
async def get_thing(id: str, session: AsyncSession = Depends(get_async_session)):
    thing = await session.get(Thing, id)
container/src/main.py

For this method, we’re handling GET requests to the /things route. This function gets a dependency injected into it, which is the session for our database, which we’ll need to run the query to get the rows.

We use the .execute method on the session object to run a select query on the Thing object. Minus any other parameters, this will just get everything in the table.

Finally, we build a list of ThingModels to return using the fetched rows. The .scalars() method we call on our things object that was returned by the query is what actually populates the results with data.

GET One

Next, we’re going to define how to get one row from the database:

@app.get("/things")
async def get_things(session: AsyncSession = Depends(get_async_session)) -> typing.List[ThingModel]:
    things = await session.execute(select(Thing))
    return [ThingModel(id=thing.id, text=thing.text) for thing in things.scalars()]

@app.get("/things/{id}")
async def get_thing(id: str, session: AsyncSession = Depends(get_async_session)):
    thing = await session.get(Thing, id)
    if thing:
        return ThingModel(id=thing.id, text=thing.text)

@app.post("/things")
async def post_thing(thing: ThingModel, session: AsyncSession = Depends(get_async_session)):
    session.add(Thing(text=thing.text))
    await session.commit()
container/src/main.py

For this method, we’re handling GET requests to the /things/{id} route. The id in curly braces means it’s an URL parameter – it’ll change depending on which thing we’re talking about.

The arguments for this route are the id parameter from the URL and the session that will be injected for us to use to get our row from the database.

We first wget a thing by calling the session’s .get method, which takes the table class and the id of the row we want. Since the ID was passed as a parameter in the URL, we just pass it to the .get method as-is.

We then check to make sure we actually got something before attempting to build a ThingModel to the user. This is an important step because there’s nothing stopping a user from directly accessing the API URL for a thing that doesn’t exist! If we didn’t have this check in there, we’d get a stack trace.

POST

Next we define the route that will actually create new things:

    if thing:
        return ThingModel(id=thing.id, text=thing.text)

@app.post("/things")
async def post_thing(thing: ThingModel, session: AsyncSession = Depends(get_async_session)):
    session.add(Thing(text=thing.text))
    await session.commit()

@app.delete("/things/{id}")
async def delete_thing(id: str, session: AsyncSession = Depends(get_async_session)):
    thing = await session.get(Thing, id)
    await session.delete(thing)
container/src/main.py

Here, we’re handling POST requests to the /things route. This function takes a POST‘ed thing in the form of a ThingModel which will be automatically populated with data using the JSON that is in the client’s request body. This is very convenient compared to how we used to have to do things like this. Like all the other routes, this one also gets a session object injected into it.

In the body of the function, we are adding a new Thing (remember, this is the SQLAlchemy model, not the Pydantic one) to the database using the session‘s .add method.

We then have to commit the changes by calling session.commit, which will tell SQLAlchemy to go ahead and make those changes. This extra step is a safeguard that most ORM libraries use in case an error happens halfway through making our changes to the data, so we don’t make any changes at all unless everything works properly.

DELETE

Now we define our route to delete a thing from the database:

    session.add(Thing(text=thing.text))
    await session.commit()

@app.delete("/things/{id}")
async def delete_thing(id: str, session: AsyncSession = Depends(get_async_session)):
    thing = await session.get(Thing, id)
    await session.delete(thing)
    await session.commit()

@app.put("/things/{id}")
async def put_thing(id: str, new_thing: ThingModel, session: AsyncSession = Depends(get_async_session)):
    thing = await session.get(Thing, id)
container/src/main.py

We’re handling DELETE requests to the /things/{id} route. Just like the GET request to get one thing form the database, we have the thing‘s id is a URL parameter here.

We pull it out of the database the same way we did before, with the session‘s .get method. Then, we call .delete on the thing we pulled out, and then .commit to make the changes stick.

PUT

Finally, we’ll define the route to update a thing in the database:

    await session.commit()

@app.put("/things/{id}")
async def put_thing(id: str, new_thing: ThingModel, session: AsyncSession = Depends(get_async_session)):
    thing = await session.get(Thing, id)
    thing.text = new_thing.text
    await session.commit()

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8080, reload=True)
container/src/main.py

We’re handling PUT requests to the /things/{id} route here. Just like before, we get the id of the thing we want from the URL parameter {id}, which is passed as an argument to the function. We also accept a new thing as an argument (an instance of ThingModel populated from the request body’s JSON) that contains the new text field that we’ll update the thing with.

We get the thing out of the database just like the GET and DELETE methods, then just set the text property to the text property of the new thing.

Finally, we commit the changes in the database.

Start your FastAPI App

We will be starting our FastAPI app using uvicorn programatically:

    thing.text = new_thing.text
    await session.commit()

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8080, reload=True)
container/src/main.py

We tell it the file and name of the app object to use, which host we want to listen on (0.0.0.0 means all hosts, including localhost), which port, and that we want to use hot reloading to automatically restart the server when we change the code. This is useful for rapid development.

Check Your Work

That’s all there is to creating a rudimentary CRUD API for one table in a SQLite database using SQLAlchemy and FastAPI. Pretty simple.

Now, we can start it up with docker compose up and you should see some output like below:

docker compose up
[+] Running 2/2
 ⠿ Network fastapi-sqlalchemy-sqlite-example_default  Created                                                                                             0.1s
 ⠿ Container fastapi-sqlalchemy-sqlite-example-api-1  Created                                                                                             0.1s
Attaching to fastapi-sqlalchemy-sqlite-example-api-1
fastapi-sqlalchemy-sqlite-example-api-1  | INFO:     Will watch for changes in these directories: ['/opt/container']
fastapi-sqlalchemy-sqlite-example-api-1  | INFO:     Uvicorn running on http://0.0.0.0:8080 (Press CTRL+C to quit)
fastapi-sqlalchemy-sqlite-example-api-1  | INFO:     Started reloader process [1] using WatchFiles
fastapi-sqlalchemy-sqlite-example-api-1  | INFO:     Started server process [8]
fastapi-sqlalchemy-sqlite-example-api-1  | INFO:     Waiting for application startup.
fastapi-sqlalchemy-sqlite-example-api-1  | INFO:     Application startup complete.

Then, in your web browser, visit http://localhost:8080/docs to view the built-in Swagger API documentation that comes with FastAPI:

Create Some Rows

We’ll create some rows using the POST section:

I went ahead and created 3 with the default value of “string” for the text field. This doesn’t matter, since that field does not need to be unique.

Get All Rows

Next, let’s make sure those rows were created by fetching them all with the GET /things route:

Great, looks like they’re all there.

Get a Single Row

Now, we’ll get one row using the GET /things/{id} route:

I put in 2 as the id to get that thing row, and it looks like it worked.

Delete a Row

Now, we’ll DELETE the thing with id of 2 using the DELETE /things/{id} route:

Nothing gets returned here to indicate success except a 200 response code. So, let’s fetch all things again to make sure thing 2 doesn’t exist anymore:

Looks like it worked!

Update a Row

Finally, we’ll update a row using the PUT method on /things/{id}:

Then check to see if our “HELLO WORLD” message is reflected in thing 3 when we fetch all rows again:

Success!

Conclusion

Creating an API backed by a database couldn’t be simpler when using FastAPI and SQLAlchemy together. We covered how to implement a basic CRUD API with this technology stack, and how to use the built-in Swagger docs that come with FastAPI to test it out.

In a real app, the next step would be to create a front-end for this with something like React, Vue, or Angular.