Using sqlalchemy with FastAPI

using sqlalchemy with FastAPI

Prerequisites

Install the following packages using pip

pip install fastapi sqlalchemy psycopg2-binary uvicorn

Create fastapi app

  • create a file main.py under src directory.
  • Let's add the below code to create the app.

main.py

from fastapi import FastAPI

app = FastAPI()
  • Now, change directory with command cd src and run the fastapi with command uvicorn main:app --reaload.
  • We can able to see the API docs at http://localhost:8000/docs

Database connection with sqlalchemy

  • create file src/db.py

Create database engine

  • add below code to create database engine
from sqlalchemy import create_engine

db_url = 'postgresql://awesome_user:secret@localhost:5432/awesome_store'

engine = create_engine(db_url)
  • we need to pass database url to function create_engine. It should be in the format driver://username:password@hostname:port/database_name

Create session using database engine

  • Create session class using sessionmaker using sqlalchemy
  • Add a utility function get_session to get the db session object.
  • By using session, we can execute the SQL and sqlalchemy ORM queries in the database.

src/db.py

from contextlib import contextmanager
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

db_url = 'postgresql://postgres:root@localhost:5432/awesome_store'
engine = create_engine(db_url)

LocalSession = sessionmaker(bind=engine)

@contextmanager
def get_session():
    session = LocalSession()
    try:
        yield session
    finally:
        session.close()

Create sqlalchemy ORM table

  • Let's say we have below table with name category in the database.
id name description
1 Bread staple food
2 Fruits Fruits
  • Now, create a sqlalchemy table/model object. To do that let's create src/models.py and below code.

src/models.py

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import INTEGER, VARCHAR, Column

Base = declarative_base()

class Category(Base):
    __tablename__ = 'category'
    id = Column('id', INTEGER, primary_key=True)
    name = Column('name', VARCHAR(30), nullable=False)
    description = Column('description', VARCHAR(255))

Add sqlalchemy queries

  • By using the ORM table/model we can write the queries.
  • Let's write the below query to fetch all categories
  • Create file src/queries.py and below code.
from sqlalchemy import select
from models import Category
from db import get_session

def get_categories():
  stmt = select(Category).filter()
  with get_session() as session:
    result = session.execute(stmt)
    return result.scalars().all()

Add GET api endpoint "/api/categories"

  • Let's add an api endpoint /api/categories to the fastapi app
  • To do that, open src/main.py and update the code as follows.

src/main.py

from fastapi import FastAPI
from queries import get_categories

app = FastAPI()

@app.get('/api/categories')
def get_categories_list():
  return get_categories()
  • Now access the above endpoint to see the data in the table. Find the sample response below.
[
  {
    "id": 1,
    "name": "Bread",
    "description": "staple food"
  },
  {
    "id": 2,
    "name": "Fruits",
    "description": "Fruits"
  }
]

Detailed video

References

Related Blog Articles

Alembic Ddl Operations

We can perform the DDL operations on the databases with alembic without writing the custom SQL in most of the cases and maintain the revision history.

Getting Started With Fastapi

FastAPI framework, high performance, easy to learn, fast to code, ready for production.