Skip to main content

SQLAlchemy

What is a SQLAlchemy​

  • One on the most popular python ORMs.
  • It's a standalone library.
  • It can be used with any other python web frameworks or python based applications
caution

SQLAlchemy does not have an underlining driver. Remember to install the appropriate database driver.

Connect FastAPI

FastAPI with SQLAlchemy

How to connect FastAPI to SQLAlchemy​

File structure​

info

The file __init__.py is just an empty file, but it tells Python that sql_app with all its modules (Python files) is a package.

└── sql_app
    ├── __init__.py
    ├── crud.py
    ├── database.py
    ├── main.py
    ├── models.py
    └── schemas.py

Step 1: Create a database URL for SQLAlchemy​

This grants access to the database server.

SQLALCHEMY_DATABASE_URL = "postgresql://<username>:<password>@<ip-address/hostname>/<database_name>"

Step 2: Create the SQLAlchemy engine​

Responsible for establishing a connection to Postgres.

sql_app/main.py
from sqlalchemy import create_engine

SQLALCHEMY_DATABASE_URL = "postgresql://<username>:<password>@<postgresserver>/<database_name>"

engine = create_engine(SQLALCHEMY_DATABASE_URL)

Step 3: Create a SessionLocal class​

Responsible for communicating with the database

sql_app/main.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "postgresql://<username>:<password>@<ip-address/hostname>/<database_name>"

engine = create_engine(SQLALCHEMY_DATABASE_URL)

SessionLocal = sessionmaker(autocommit=False)

Step 4: Create a Base class​

The Base class to create each of the database models or classes (the ORM models)

sql_app/main.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "postgresql://<username>:<password>@<ip-address/hostname>/<database_name>"

engine = create_engine(SQLALCHEMY_DATABASE_URL)

SessionLocal = sessionmaker(autocommit=False)

Base = declarative_base()

Step 5: Create the database models​

Create a file named models.py

└── venv
└── sql_app
    ├── __init__.py
    ├── crud.py
    ├── database.py
    ├── main.py
    ├── models.py
    └── schemas.py
To run the application use 👇
  uvicorn app.main:app --reload
Tip

SQLAlchemy uses the term "model" to refer to these classes and instances that interact with the database.

But Pydantic also uses the term "model" to refer to something different, the data validation, conversion, and documentation classes and instances.

sql_app/models.py
from .database import Base


class User(Base):
    __tablename__ = "users"

class Item(Base):
    __tablename__ = "items"

Step 6: Create model attributes/columns​

Add the items attributes

sql_app/models.py
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from .database import Base

class User(Base):
    __tablename__ = "users"

    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)

class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

Step 7: Create a dependency​

To have an independent database session/connection (SessionLocal) per request, use the same session through all the request and then close it after the request is finished.

sql_app/main.py
from typing import List

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

CRUD Methods​

GET itmes​

sql_app/main.py
from fastapi import FastAPI, Depends
from .database import engine
import psycopg2

from . import models
from sqlalchemy.orm import Session
from .database import engine, get_db

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

@app.get("/table_name")
def get_table_name(db: Session = Depends(get_db)):
    table_name = db.query(models.Tablename).all()
    return {"data": table_name}

GET single item​

sql_app/main.py
from fastapi import FastAPI, Depends
from .database import engine
from . import models, schemas
import psycopg2

from . import models
from sqlalchemy.orm import Session
from .database import engine, get_db

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

@app.get("/table_name/{item.id}")
def create_item(item_id: int, db: Session = Depends(get_db)):
  db.add(sqlAchemy.models.TableItem(**item.dict()))
  db.commit()
  return {"status": "success"}

CREATE item​

When the endpoint is called we'll hit the create_item function triggering a session. In this session we'll check the request for a body. THan we'll create a new item for the table than we'll (permenantly) write to the db.

sql_app/main.py
from fastapi import FastAPI, Depends
from .database import engine
from . import models, schemas
import psycopg2

from . import models
from sqlalchemy.orm import Session
from .database import engine, get_db

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

@app.post("/table_name")
def create_item(item: pydantic.Schema, db: Session = Depends(get_db)):
  db.add(sqlAchemy.models.TableItem(**item.dict()))
  db.commit()
  return {"status": "success"}