SQLAlchemy layer
- Python course
- Advanced
- Lesson 81/83
- 6 min
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.
Practice 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())).
Starter code
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())
# bring in create_engine, Column, Integer, String, StaticPool from sqlalchemy
# bring in declarative_base and sessionmaker from sqlalchemy.orm
# define Base, write the Task model (id, title columns)
# create the engine (in-memory sqlite, StaticPool, check_same_thread=False)
# create tables, open a session, add two Tasks and commit
# redefine fetch_tasks to query the database
# run fetch_tasks with asyncio.run and print the resultSample 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()))Expected output
[{'id': 1, 'title': 'Görev 1'}, {'id': 2, 'title': 'Görev 2'}]
Questions and answers
Review the lesson questions together with their answers. To test yourself, solve the lesson interactively.
1 What is needed when using in-memory SQLite with FastAPI TestClient?
- No extra settings
- StaticPool and check_same_thread=False Correct answer
- Only StaticPool
- Only check_same_thread=False
Explanation: StaticPool shares the connection; check_same_thread=False allows the thread.
2 Complete the table name
class Task(Base):
____ = "tasks"class Task(Base):
__tablename__ = "tasks"Explanation: __tablename__ sets the table name.
3 Match the SQLAlchemy concepts
| Term | Meaning |
|---|---|
create_engine |
Creates the engine |
declarative_base |
Model base |
Column |
Defines a column |
session.query |
Runs a query |