Python
SQLAlchemy layer
Write database operations with SQLAlchemy.
SQLAlchemy layer
The third layer replaces the fixed list with a real database query. With SQLAlchemy we set up an in-memory SQLite, define a Task model, and query with session.query(Task).all().
from sqlalchemy import create_engine, Column, Integer, String, StaticPool
from sqlalchemy.orm import declarative_base, sessionmaker
engine = create_engine("sqlite:///:memory:", poolclass=StaticPool)
Without StaticPool, in-memory SQLite creates a fresh database per connection. Since FastAPI's TestClient uses a separate thread, check_same_thread=False and StaticPool are required to share the connection.
The Task model
A SQLAlchemy model is a class derived from declarative_base(). __tablename__ sets the table name; Column defines the columns.
class Task(Base):
__tablename__ = "tasks"
id = Column(Integer, primary_key=True)
title = Column(String)
This class is a bridge between Python objects and database rows. session.add(Task(...)) adds a new row; session.commit() persists it. Queries go from the class name to the table automatically.
Exercise Add the Task model and database query
The starting code contains the previous lesson's fetch_tasks and fetch_all solutions. Add the following:
from sqlalchemy import create_engine, Column, Integer, String, StaticPoolfrom sqlalchemy.orm import declarative_base, sessionmakerBase = declarative_base()class Task(Base): — __tablename__ = "tasks", id = Column(Integer, primary_key=True), title = Column(String)- Create the engine:
engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool) Base.metadata.create_all(engine), Session = sessionmaker(bind=engine), session = Session()- Add two Tasks and commit
- Redefine
fetch_tasks: return [{"id": t.id, "title": t.title} for t in session.query(Task).all()]
Then print the result with print(asyncio.run(fetch_tasks())).
Show example solution
import asyncio
async def fetch_tasks():
return [{"id": 1, "title": "Görev 1"}, {"id": 2, "title": "Görev 2"}]
async def fetch_all():
return await asyncio.gather(fetch_tasks(), fetch_tasks())
from sqlalchemy import create_engine, Column, Integer, String, StaticPool
from sqlalchemy.orm import declarative_base, sessionmaker
Base = declarative_base()
class Task(Base):
__tablename__ = "tasks"
id = Column(Integer, primary_key=True)
title = Column(String)
engine = create_engine(
"sqlite:///:memory:",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
session.add(Task(id=1, title="Görev 1"))
session.add(Task(id=2, title="Görev 2"))
session.commit()
async def fetch_tasks():
return [{"id": t.id, "title": t.title} for t in session.query(Task).all()]
print(asyncio.run(fetch_tasks()))
Test yourself
-
What is needed when using in-memory SQLite with FastAPI TestClient?
- No extra settings
- StaticPool and check_same_thread=False
- Only StaticPool
- Only check_same_thread=False
Show answer
StaticPool and check_same_thread=False — StaticPool shares the connection; check_same_thread=False allows the thread.
-
Complete the table name
class Task(Base):
____ = "tasks"
Show answer
class Task(Base):
__tablename__ = "tasks"
-
Match the SQLAlchemy concepts
- create_engine
- Creates the engine
- declarative_base
- Model base
- Column
- Defines a column
- session.query
- Runs a query
Next lesson: FastAPI layer
