sqlalchemy and alembic integration in FastAPI

Question 1

I am building a fastAPI + sqlalchemy + alembic + docker-compose reusable template.Full source code:https://github.com/mascai/fastapi_template (commit 77ce7f2)

I would appreciate pull-requests =)

The project is working but I worry about alembic and sqlalchemy integration

The first point:

I have to import users (orm model) in theinit.py file, alembic doesn't see changes in the user model without this import:

# app/models/__init__.pyfrom sqlalchemy.ext.declarative import declarative_baseBase = declarative_base()from . import users # IMPORTANT: intended to register models for alembic (should be at the end of the file), alembic revision --autogenerate -m "migration name"

I am creating migrations manually like this: (What is the best practise to create and apply migrations?)

docker exec -ti fastapi-container-name bashalembic revision --autogenerate -m "migration name"alembic upgrade head

What is the proper way to link sqlalchemy and alembic in my code?

Project tree and main files:

├── app│   ├── Dockerfile│   ├── __init__.py│   ├── alembic│   │   ├── README│   │   ├── env.py│   │   ├── script.py.mako│   │   └── versions│   │       └── 486ef6640756_initial_commit.py│   ├── alembic.ini│   ├── api│   │   └── v1│   │       ├── __init__.py│   │       └── users.py│   ├── database│   │   ├── __init__.py│   │   └── session.py│   ├── main.py│   ├── models│   │   ├── __init__.py│   │   └── users.py│   ├── requirements.txt│   ├── schemas│   │   ├── __init__.py│   │   └── users.py│   └── utils│       └── __init__.py├── docker-compose.yaml

Alembic settings:

# /app/alembic/env.pyfrom logging.config import fileConfigfrom sqlalchemy import engine_from_configfrom sqlalchemy import poolfrom alembic import context# this is the Alembic Config object, which provides# access to the values within the .ini file in use.config = context.config# Interpret the config file for Python logging.# This line sets up loggers basically.if config.config_file_name is not None:    fileConfig(config.config_file_name)# add your model's MetaData object here# for 'autogenerate' support# from myapp import mymodel# target_metadata = mymodel.Base.metadatafrom app.models import Basetarget_metadata = Base.metadatafrom app.database.session import DATABASE_URLconfig.set_main_option('sqlalchemy.url', DATABASE_URL)...

Main file

# app/main.pyfrom fastapi import FastAPI, APIRouter, Depends, HTTPExceptionfrom sqlalchemy.orm import Sessionfrom typing import List, Dict, Anyfrom models.users import Userfrom schemas.users import UserSchema, BaseUserSchemafrom database.session import get_dbfrom api.v1.users import users_routerapp = FastAPI()app.include_router(users_router)@app.get("/health_check")def health_check():    return {"Hello": "World"}

Model description

# app/models.pyfrom sqlalchemy import Column, Integer, Stringfrom models import Baseclass User(Base):    __tablename__ = 'users'    id = Column(Integer, primary_key=True)    name = Column(String, unique=True)    age = Column(Integer)

Api example

app/api/v1/users.pyfrom fastapi import APIRouter, Depends, HTTPExceptionfrom sqlalchemy.orm import Sessionfrom typing import List, Dict, Anyfrom models.users import Userfrom schemas.users import UserSchema, BaseUserSchemafrom database.session import get_dbusers_router = APIRouter()@users_router.get("/users", tags=["user"])def get_users(db: Session = Depends(get_db)):    return db.query(User).all()@users_router.post("/users", response_model=UserSchema, tags=["user"])def create_user(user_data: BaseUserSchema, db: Session = Depends(get_db)):    query = db.query(User).filter(User.name == user_data.name).first()    if query:        raise HTTPException(status_code=400, detail="User already exist")    new_user = User(name=user_data.name, age=user_data.age)    db.add(new_user)    db.commit()    return new_user

docker-compose.yaml

version: '3.8'services:  backend-api:    build:      context: ./app      dockerfile: Dockerfile    ports:      - "8080:8080"    command: uvicorn main:app --host 0.0.0.0 --port 8080  --reload    env_file: ".env"    volumes:    - ./app/alembic/versions/:/app/alembic/versions/    depends_on:      - db  db:    image: postgres:latest    ports:      - "5432:5432"    expose:      - 5432    env_file: ".env"    volumes:      - pgdata:/var/lib/postgresql/data volumes:  pgdata:

Question 2

Looks good.I assume you're using sqlalchemy > 2.

base

Base = declarative_base()from . import users

Maybe you would find it more convenient to defineBasein theusers.py module?

(Also, an orthogonal religious war detail:Many practitioners prefer to give a singular noun nameto each table, e.g. CREATE TABLE USER.It's already obvious a table can hold multiple rows.I suspect that if USER wasn't a reserved SQL keywordwe would see fewer CREATE TABLE USERS statements.)

I'm not saying your code is wrong,and I realize you probably aren't type checking.But I could never getmypy --strict toaccept the OPBase = assignment.I always define it in a linter friendly way like this:

from sqlalchemy.orm import DeclarativeBaseclass Base(DeclarativeBase):    """Base class for all tables."""

migrations

alembic revision --autogenerate -m "migration name"alembic upgrade head

That looks like a perfectly good way to manually create migrations;I see nothing that should be revised.

Similar to typing$ git status && git log -2 pretty frequently,before and after a migration I tend to run commandslike$ alembic current && alembic history more than once,just to remain aware of the present status.

What is the proper way to link sqlalchemy and alembic in my code?

Similarly, your tree structure looks fine.It's important for alembic to have its own bit of namespace,its ownalembic/ folder, and you have that.

boring function name

@app.get("/health_check")def health_check():

Excellent!Thank you for being boring.Using the identical spelling for the URL and for the function nameis exactly the right approach.(Well, fine,"/health-check" would also correspond tohealth_check, whatever.)

There's room for creativity in some parts of the code,but this isn't one of them; you are doing the Right Thing.

Thedef get_users is maybe OK, and it might evensuggestdef get_health_check.It's not yet obvious to me if we always do a PK lookup onidand so we return a sequence of either 0 or 1 users,that is, never multiple users.If max results is 1,then consider naming the URL/user,and perhaps returning 404 when zero users are found.

The singulardef create_user is rather awkward.Consider fixing it by changing its URL spelling to/user.

optional value

class User(Base):    ...    age = Column(Integer)

Depending on your business use case, you might possiblywish to specify, nullable=False) for that column.

In general, it's easier to reason about relations whenthere's fewer optional attributes.And it's much easier to insist on "NOT NULL" from the get go,than to tack it on later in a subsequent PR,where you have to worry about "what established thingswill this possibly break?".

old-style annotations

from typing import List, Dict, Any

Prefer to pull in justAny.Code written for modern interpreters (certainly for 3.10+)should just specify e.g.list[Any] anddict[str, Any]in signatures.(Or more specific thanAny, such asstr, where applicable.)

DB server

I guess this works fine for you:

    volumes:      - pgdata:/var/lib/postgresql/data

Me? I would worry about accidentally havingtwo simultaneous servers fight over the data,and a hypervisor not exposing flock() detailsto tell the latecomer "you lose!".

Consider creating a separate container fora central postgres server which yourvarious microservices connect to via TCP.

BTW, since the OP does not (yet) use any vendor-specificfeatures, you have the flexibility to switch to sqlitejust by changing a connect string.Postgres is a wonderful product that performs well under loadand offers many unique fancy features.But for a vendor-agnostic app such as this one,it turns out that automated integrationteststhat do e2e sqlite transactions are really quite convenient.They let you exercise nearly 100% of the application target codewith very low testing effort.And when you're done, a simple$ rm /tmp/test.sqliteresets things back to square one for the next test run.

Question 3

Consider creating a separate container for a central postgres server... - could you elaborate? That's exactly what happens here, a separate postgres container to which the (monolithic, not sure why you mention microservices) backend talks over TCP. And compose volumes are prefixed with the project name so nobody will "fight over the data", that data is exclusively used by one postgres container...

Question 4

Ok, good! I failed to understand that global context upon my initial reading.

J_H 43.3k3 gold badges38 silver badges158 bronze badges · Answer 1 · 2025-02-02 17:44:53Z

Looks good.I assume you're using sqlalchemy > 2.

base

Base = declarative_base()from . import users

Maybe you would find it more convenient to defineBasein theusers.py module?

(Also, an orthogonal religious war detail:Many practitioners prefer to give a singular noun nameto each table, e.g. CREATE TABLE USER.It's already obvious a table can hold multiple rows.I suspect that if USER wasn't a reserved SQL keywordwe would see fewer CREATE TABLE USERS statements.)

I'm not saying your code is wrong,and I realize you probably aren't type checking.But I could never getmypy --strict toaccept the OPBase = assignment.I always define it in a linter friendly way like this:

from sqlalchemy.orm import DeclarativeBaseclass Base(DeclarativeBase):    """Base class for all tables."""

migrations

alembic revision --autogenerate -m "migration name"alembic upgrade head

That looks like a perfectly good way to manually create migrations;I see nothing that should be revised.

Similar to typing$ git status && git log -2 pretty frequently,before and after a migration I tend to run commandslike$ alembic current && alembic history more than once,just to remain aware of the present status.

What is the proper way to link sqlalchemy and alembic in my code?

Similarly, your tree structure looks fine.It's important for alembic to have its own bit of namespace,its ownalembic/ folder, and you have that.

boring function name

@app.get("/health_check")def health_check():

Excellent!Thank you for being boring.Using the identical spelling for the URL and for the function nameis exactly the right approach.(Well, fine,"/health-check" would also correspond tohealth_check, whatever.)

There's room for creativity in some parts of the code,but this isn't one of them; you are doing the Right Thing.

Thedef get_users is maybe OK, and it might evensuggestdef get_health_check.It's not yet obvious to me if we always do a PK lookup onidand so we return a sequence of either 0 or 1 users,that is, never multiple users.If max results is 1,then consider naming the URL/user,and perhaps returning 404 when zero users are found.

The singulardef create_user is rather awkward.Consider fixing it by changing its URL spelling to/user.

optional value

class User(Base):    ...    age = Column(Integer)

Depending on your business use case, you might possiblywish to specify, nullable=False) for that column.

In general, it's easier to reason about relations whenthere's fewer optional attributes.And it's much easier to insist on "NOT NULL" from the get go,than to tack it on later in a subsequent PR,where you have to worry about "what established thingswill this possibly break?".

old-style annotations

from typing import List, Dict, Any

Prefer to pull in justAny.Code written for modern interpreters (certainly for 3.10+)should just specify e.g.list[Any] anddict[str, Any]in signatures.(Or more specific thanAny, such asstr, where applicable.)

DB server

I guess this works fine for you:

    volumes:      - pgdata:/var/lib/postgresql/data

Me? I would worry about accidentally havingtwo simultaneous servers fight over the data,and a hypervisor not exposing flock() detailsto tell the latecomer "you lose!".

Consider creating a separate container fora central postgres server which yourvarious microservices connect to via TCP.

BTW, since the OP does not (yet) use any vendor-specificfeatures, you have the flexibility to switch to sqlitejust by changing a connect string.Postgres is a wonderful product that performs well under loadand offers many unique fancy features.But for a vendor-agnostic app such as this one,it turns out that automated integrationteststhat do e2e sqlite transactions are really quite convenient.They let you exercise nearly 100% of the application target codewith very low testing effort.And when you're done, a simple$ rm /tmp/test.sqliteresets things back to square one for the next test run.

Consider creating a separate container for a central postgres server... - could you elaborate? That's exactly what happens here, a separate postgres container to which the (monolithic, not sure why you mention microservices) backend talks over TCP. And compose volumes are prefixed with the project name so nobody will "fight over the data", that data is exclusively used by one postgres container...
Ok, good! I failed to understand that global context upon my initial reading.

Movatterモバイル変換

Stack Exchange Network

sqlalchemy and alembic integration in FastAPI

1 Answer1

base

migrations

boring function name

optional value

old-style annotations

DB server

You mustlog in to answer this question.

Related

Hot Network Questions

Subscribe to RSS