Skip to content

SQL (Relational) Databases

FastAPI doesn't require you to use a SQL (relational) database.

But you can use any relational database that you want.

Here we'll see an example using SQLAlchemy.

You can easily adapt it to any database supported by SQLAlchemy, like:

  • PostgreSQL
  • MySQL
  • SQLite
  • Oracle
  • Microsoft SQL Server, etc.

In this example, we'll use PostgreSQL.

Note

Notice that most of the code is the standard SQLAlchemy code you would use with any framework.

The FastAPI specific code is as small as always.

Import SQLAlchemy components

For now, don't pay attention to the rest, only the imports:

from fastapi import FastAPI

from sqlalchemy import Boolean, Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base, declared_attr
from sqlalchemy.orm import scoped_session, sessionmaker

# SQLAlchemy specific code, as with any other app
SQLALCHEMY_DATABASE_URI = "postgresql://user:password@postgresserver/db"

engine = create_engine(SQLALCHEMY_DATABASE_URI, convert_unicode=True)
db_session = scoped_session(
    sessionmaker(autocommit=False, autoflush=False, bind=engine)
)


class CustomBase:
    # Generate __tablename__ automatically
    @declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()


Base = declarative_base(cls=CustomBase)


class User(Base):
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean(), default=True)


def get_user(username, db_session):
    return db_session.query(User).filter(User.id == username).first()


# FastAPI specific code
app = FastAPI()


@app.get("/users/{username}")
def read_user(username: str):
    user = get_user(username, db_session)
    return user

Define the database

Define the database that SQLAlchemy should connect to:

from fastapi import FastAPI

from sqlalchemy import Boolean, Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base, declared_attr
from sqlalchemy.orm import scoped_session, sessionmaker

# SQLAlchemy specific code, as with any other app
SQLALCHEMY_DATABASE_URI = "postgresql://user:password@postgresserver/db"

engine = create_engine(SQLALCHEMY_DATABASE_URI, convert_unicode=True)
db_session = scoped_session(
    sessionmaker(autocommit=False, autoflush=False, bind=engine)
)


class CustomBase:
    # Generate __tablename__ automatically
    @declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()


Base = declarative_base(cls=CustomBase)


class User(Base):
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean(), default=True)


def get_user(username, db_session):
    return db_session.query(User).filter(User.id == username).first()


# FastAPI specific code
app = FastAPI()


@app.get("/users/{username}")
def read_user(username: str):
    user = get_user(username, db_session)
    return user

Tip

This is the main line that you would have to modify if you wanted to use a different database than PostgreSQL.

Create the SQLAlchemy engine

from fastapi import FastAPI

from sqlalchemy import Boolean, Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base, declared_attr
from sqlalchemy.orm import scoped_session, sessionmaker

# SQLAlchemy specific code, as with any other app
SQLALCHEMY_DATABASE_URI = "postgresql://user:password@postgresserver/db"

engine = create_engine(SQLALCHEMY_DATABASE_URI, convert_unicode=True)
db_session = scoped_session(
    sessionmaker(autocommit=False, autoflush=False, bind=engine)
)


class CustomBase:
    # Generate __tablename__ automatically
    @declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()


Base = declarative_base(cls=CustomBase)


class User(Base):
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean(), default=True)


def get_user(username, db_session):
    return db_session.query(User).filter(User.id == username).first()


# FastAPI specific code
app = FastAPI()


@app.get("/users/{username}")
def read_user(username: str):
    user = get_user(username, db_session)
    return user

Create a scoped_session

from fastapi import FastAPI

from sqlalchemy import Boolean, Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base, declared_attr
from sqlalchemy.orm import scoped_session, sessionmaker

# SQLAlchemy specific code, as with any other app
SQLALCHEMY_DATABASE_URI = "postgresql://user:password@postgresserver/db"

engine = create_engine(SQLALCHEMY_DATABASE_URI, convert_unicode=True)
db_session = scoped_session(
    sessionmaker(autocommit=False, autoflush=False, bind=engine)
)


class CustomBase:
    # Generate __tablename__ automatically
    @declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()


Base = declarative_base(cls=CustomBase)


class User(Base):
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean(), default=True)


def get_user(username, db_session):
    return db_session.query(User).filter(User.id == username).first()


# FastAPI specific code
app = FastAPI()


@app.get("/users/{username}")
def read_user(username: str):
    user = get_user(username, db_session)
    return user

Very Technical Details

Don't worry too much if you don't understand this. You can still use the code.

This scoped_session is a feature of SQLAlchemy.

The resulting object, the db_session can then be used anywhere a a normal SQLAlchemy session.

It can be used as a global because it is implemented to work independently on each "thread", so the actions you perform with it in one path operation function won't affect the actions performed (possibly concurrently) by other path operation functions.

Create a CustomBase model

This is more of a trick to facilitate your life than something required.

But by creating this CustomBase class and inheriting from it, your models will have automatic __tablename__ attributes (that are required by SQLAlchemy).

That way you don't have to declare them explicitly.

So, your models will behave very similarly to, for example, Flask-SQLAlchemy.

from fastapi import FastAPI

from sqlalchemy import Boolean, Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base, declared_attr
from sqlalchemy.orm import scoped_session, sessionmaker

# SQLAlchemy specific code, as with any other app
SQLALCHEMY_DATABASE_URI = "postgresql://user:password@postgresserver/db"

engine = create_engine(SQLALCHEMY_DATABASE_URI, convert_unicode=True)
db_session = scoped_session(
    sessionmaker(autocommit=False, autoflush=False, bind=engine)
)


class CustomBase:
    # Generate __tablename__ automatically
    @declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()


Base = declarative_base(cls=CustomBase)


class User(Base):
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean(), default=True)


def get_user(username, db_session):
    return db_session.query(User).filter(User.id == username).first()


# FastAPI specific code
app = FastAPI()


@app.get("/users/{username}")
def read_user(username: str):
    user = get_user(username, db_session)
    return user

Create the SQLAlchemy Base model

from fastapi import FastAPI

from sqlalchemy import Boolean, Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base, declared_attr
from sqlalchemy.orm import scoped_session, sessionmaker

# SQLAlchemy specific code, as with any other app
SQLALCHEMY_DATABASE_URI = "postgresql://user:password@postgresserver/db"

engine = create_engine(SQLALCHEMY_DATABASE_URI, convert_unicode=True)
db_session = scoped_session(
    sessionmaker(autocommit=False, autoflush=False, bind=engine)
)


class CustomBase:
    # Generate __tablename__ automatically
    @declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()


Base = declarative_base(cls=CustomBase)


class User(Base):
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean(), default=True)


def get_user(username, db_session):
    return db_session.query(User).filter(User.id == username).first()


# FastAPI specific code
app = FastAPI()


@app.get("/users/{username}")
def read_user(username: str):
    user = get_user(username, db_session)
    return user

Create your application data model

Now this is finally code specific to your app.

Here's a user model that will be a table in the database:

from fastapi import FastAPI

from sqlalchemy import Boolean, Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base, declared_attr
from sqlalchemy.orm import scoped_session, sessionmaker

# SQLAlchemy specific code, as with any other app
SQLALCHEMY_DATABASE_URI = "postgresql://user:password@postgresserver/db"

engine = create_engine(SQLALCHEMY_DATABASE_URI, convert_unicode=True)
db_session = scoped_session(
    sessionmaker(autocommit=False, autoflush=False, bind=engine)
)


class CustomBase:
    # Generate __tablename__ automatically
    @declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()


Base = declarative_base(cls=CustomBase)


class User(Base):
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean(), default=True)


def get_user(username, db_session):
    return db_session.query(User).filter(User.id == username).first()


# FastAPI specific code
app = FastAPI()


@app.get("/users/{username}")
def read_user(username: str):
    user = get_user(username, db_session)
    return user

Get a user

By creating a function that is only dedicated to getting your user from a username (or any other parameter) independent of your path operation function, you can more easily re-use it in multiple parts and also add unit tests for it:

from fastapi import FastAPI

from sqlalchemy import Boolean, Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base, declared_attr
from sqlalchemy.orm import scoped_session, sessionmaker

# SQLAlchemy specific code, as with any other app
SQLALCHEMY_DATABASE_URI = "postgresql://user:password@postgresserver/db"

engine = create_engine(SQLALCHEMY_DATABASE_URI, convert_unicode=True)
db_session = scoped_session(
    sessionmaker(autocommit=False, autoflush=False, bind=engine)
)


class CustomBase:
    # Generate __tablename__ automatically
    @declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()


Base = declarative_base(cls=CustomBase)


class User(Base):
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean(), default=True)


def get_user(username, db_session):
    return db_session.query(User).filter(User.id == username).first()


# FastAPI specific code
app = FastAPI()


@app.get("/users/{username}")
def read_user(username: str):
    user = get_user(username, db_session)
    return user

Create your FastAPI code

Now, finally, here's the standard FastAPI code.

Create your app and path operation function:

from fastapi import FastAPI

from sqlalchemy import Boolean, Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base, declared_attr
from sqlalchemy.orm import scoped_session, sessionmaker

# SQLAlchemy specific code, as with any other app
SQLALCHEMY_DATABASE_URI = "postgresql://user:password@postgresserver/db"

engine = create_engine(SQLALCHEMY_DATABASE_URI, convert_unicode=True)
db_session = scoped_session(
    sessionmaker(autocommit=False, autoflush=False, bind=engine)
)


class CustomBase:
    # Generate __tablename__ automatically
    @declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()


Base = declarative_base(cls=CustomBase)


class User(Base):
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean(), default=True)


def get_user(username, db_session):
    return db_session.query(User).filter(User.id == username).first()


# FastAPI specific code
app = FastAPI()


@app.get("/users/{username}")
def read_user(username: str):
    user = get_user(username, db_session)
    return user

As we are using SQLAlchemy's scoped_session, we don't even have to create a dependency with Depends.

We can just call get_user directly from inside of the path operation function and use the global db_session.

Create the path operation function

Here we are using SQLAlchemy code inside of the path operation function, and it in turn will go and communicate with an external database.

That could potentially require some "waiting".

But as SQLAlchemy doesn't have compatibility for using await, as would be with something like:

user = await get_user(username, db_session)

...and instead we are using:

user = get_user(username, db_session)

Then we should declare the path operation without async def, just with a normal def:

from fastapi import FastAPI

from sqlalchemy import Boolean, Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base, declared_attr
from sqlalchemy.orm import scoped_session, sessionmaker

# SQLAlchemy specific code, as with any other app
SQLALCHEMY_DATABASE_URI = "postgresql://user:password@postgresserver/db"

engine = create_engine(SQLALCHEMY_DATABASE_URI, convert_unicode=True)
db_session = scoped_session(
    sessionmaker(autocommit=False, autoflush=False, bind=engine)
)


class CustomBase:
    # Generate __tablename__ automatically
    @declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()


Base = declarative_base(cls=CustomBase)


class User(Base):
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean(), default=True)


def get_user(username, db_session):
    return db_session.query(User).filter(User.id == username).first()


# FastAPI specific code
app = FastAPI()


@app.get("/users/{username}")
def read_user(username: str):
    user = get_user(username, db_session)
    return user

Migrations

Because we are using SQLAlchemy directly and we don't require any kind of plug-in for it to work with FastAPI, we could integrate database migrations with Alembic directly.

You would probably want to declare your database and models in a different file or set of files, this would allow Alembic to import it and use it without even needing to have FastAPI installed for the migrations.