Relational database interactions form the core persistence layer for web applications and backend systems. Writing raw SQL strings directly inside Python scripts leads to SQL injection vulnerabilities, lack of static type checking, and brittle schema migrations.
SQLAlchemy 2.0 is Python's premier Object Relational Mapping (ORM) framework, bridging relational database engines (PostgreSQL, SQLite, MySQL) with Python classes. Version 2.0 introduces fully typed declarative mappings (Mapped and mapped_column) and native asynchronous database sessions with asyncpg and aiosqlite.
SQLAlchemy 1.4 Legacy vs. SQLAlchemy 2.0 Modern Standards
Prerequisites & Installation
SQLAlchemy 2.0 provides fully typed declarative mappings and native async engine support.
pip install "sqlalchemy>=2.0" alembic aiosqlite
| Feature | SQLAlchemy 1.4 (Legacy) | SQLAlchemy 2.0 (Modern Standard) | Developer Benefit |
|---|---|---|---|
| Model Declarations | Column(String, primary_key=True) | mapped_column(String(50), primary_key=True) | Full Mypy / Pylance static type inference |
| Type Hints | name = Column(String) | name: Mapped[str] | Autocomplete and type safety in IDEs |
| Query Syntax | session.query(User).filter_by(name='alex') | session.scalars(select(User).where(User.name == 'alex')) | Unified typed SQL expression language |
| Asynchronous Support | Experimental plugin wrappers | Native AsyncSession & asyncpg engine | High-throughput non-blocking database queries |
Building Typed Declarative Models and Relationships
In modern SQLAlchemy 2.0, all database models inherit from DeclarativeBase. Model columns use Mapped[T] type annotations, providing seamless autocompletion in VS Code and static validation.
Relationships define foreign key associations and cascading behavior cleanly.
Complete SQLAlchemy 2.0 Database Script with Typed Queries
from typing import List
from sqlalchemy import create_engine, String, ForeignKey, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session
# 1. Declarative Base
class Base(DeclarativeBase):
pass
# 2. Database Models
class Author(Base):
__tablename__ = "authors"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(60), unique=True)
email: Mapped[str] = mapped_column(String(120), unique=True)
# 1-to-Many Relationship
articles: Mapped[List["Article"]] = relationship(back_populates="author", cascade="all, delete-orphan")
class Article(Base):
__tablename__ = "articles"
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column(String(200))
author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"))
author: Mapped["Author"] = relationship(back_populates="articles")
# 3. Initialize SQLite Database Engine
engine = create_engine("sqlite:///./production_app.db", echo=False)
Base.metadata.create_all(engine)
# 4. Insert & Query Transactions
with Session(engine) as session:
# Insert new records
author = Author(name="Alex Rivers", email="alex@ilabacademy.com")
post1 = Article(title="Async and Await in Python Guide", author=author)
post2 = Article(title="FastAPI REST API Architecture", author=author)
session.add(author)
session.commit()
# Typed 2.0 Select Statement
stmt = select(Author).where(Author.name == "Alex Rivers")
result_author = session.scalars(stmt).first()
if result_author:
print(f"Retrieved Author: {result_author.name} (Email: {result_author.email})")
print(f"Published Articles ({len(result_author.articles)}):")
for art in result_author.articles:
print(f" - {art.title}")
Database Connection Pooling & Performance Tuning
- N+1 Query Problem: When loading relationships, avoid triggering separate SQL queries for each child record. Use selectinload() or joinedload() to eagerly load relationships in a single SQL query.
- Connection Pool Sizing: In production web services (like FastAPI), configure pool_size=20 and max_overflow=10 in create_engine() to prevent connection exhaustion under high concurrency.
- Session Lifecycle: Always manage database sessions using context managers (with Session(engine) as session) to guarantee transactions are committed or rolled back automatically.
Frequently Asked Questions
Q: What is the difference between session.scalars() and session.execute() in SQLAlchemy 2.0?
A: session.execute() returns raw Row objects (tuples of columns). session.scalars() unpacks the first column directly into typed model instances, eliminating manual unpacking boilerplate.
Q: How do I perform schema migrations with SQLAlchemy 2.0?
A: Use Alembic (alembic init alembic), which automatically inspects your DeclarativeBase metadata and generates versioned SQL migration scripts.
0 Comments