PostgreSQL, MySQL & SQLAlchemy ORM
📌 Covered in this chapter:
Enterprise DB Architecture · PostgreSQL vs MySQL · The ORM Pattern · SQLAlchemy 2.0 DeclarativeBase · Mapped Columns · Session CRUD Lifecycle
Master enterprise databases and Object-Relational Mapping (ORM) in Python: client-server database comparison (PostgreSQL vs MySQL), why modern teams use ORMs, building type-safe data models with SQLAlchemy 2.0, and executing production CRUD operations with Sessions.
1Enterprise Databases: PostgreSQL vs MySQL Architecture
In large-scale production environments with millions of concurrent users, applications use dedicated Client-Server Database Clusters:
PostgreSQL ("The World's Most Advanced Open Source Relational Database"):
- Architecture: Multi-process architecture using MVCC (Multi-Version Concurrency Control), meaning readers never block writers, and writers never block readers.
- Advanced Features: Native JSONB binary indexing, geospatial data with PostGIS, vector similarity search with
pgvector(for AI and LLM embeddings), full-text search, and custom data types. - Python Drivers:
psycopg(Psycopg 3) andasyncpg(high-performance asynchronous driver).
MySQL ("The Web Standard RDBMS"):
- Architecture: Multi-threaded architecture powered by the InnoDB storage engine.
- Strengths: Optimized for high-throughput, read-heavy workloads (powers WordPress, Wikipedia, and large web portals).
- Python Drivers:
mysql-connector-pythonandPyMySQL.
💻 Reference: Standard Database Connection URLs (DSNs)
Python 3
▶ Run in Compiler
# Database Connection String (DSN) URL Anatomy:
# Format: dialect+driver://username:password@host:port/database_name
dsn_examples = {
"PostgreSQL (Standard)": "postgresql+psycopg://postgres_user:secret_pass@127.0.0.1:5432/ecommerce_db",
"PostgreSQL (Async)": "postgresql+asyncpg://postgres_user:secret_pass@127.0.0.1:5432/ecommerce_db",
"MySQL (PyMySQL)": "mysql+pymysql://db_user:secret_pass@127.0.0.1:3306/ecommerce_db",
"SQLite (Local File)": "sqlite:///production_store.db",
"SQLite (In-Memory)": "sqlite:///:memory:"
}
print("--- 🌐 Standard Database Connection URI Patterns ---")
for db_type, uri in dsn_examples.items():
print(f"• {db_type:22}: {uri}")
🔍 The 12-Factor App Rule:
Never hardcode database URLs or passwords in Python files. Always load the connection string from an environment variable: DATABASE_URL = os.getenv("DATABASE_URL").
2The ORM Paradigm: What is SQLAlchemy & Why Use It?
An Object-Relational Mapper (ORM) maps database tables to native Python classes, columns to class attributes, and table rows to class instances.
The 4 Major Advantages of an ORM:
- Type Safety & IDE Autocomplete: With modern SQLAlchemy 2.0 type annotations (
Mapped[str],Mapped[int]), your IDE provides instant autocomplete for every column and catches type bugs before runtime. - Dialect Independence: Write Python code once; SQLAlchemy translates it into PostgreSQL, MySQL, SQLite, or Oracle SQL syntax automatically.
- Automatic Migration Tooling (Alembic): As your application evolves, schema changes (adding columns, renaming tables) are tracked and applied automatically via version-controlled migration scripts.
- Native Object Lifecycle & Unit of Work: Modify an object property (
user.email = "new@mail.com") and the SQLAlchemy Session automatically tracks the modification ("dirty checking") and generates the exact SQLUPDATEstatement upon commit!
💻 Example 2: Complete SQLAlchemy 2.0 Declarative Model and Session CRUD
Python 3
▶ Run in Compiler
from typing import Optional
from sqlalchemy import create_engine, String, Float, Integer, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
# 1. Base class for all SQLAlchemy 2.0 Models:
class Base(DeclarativeBase):
pass
# 2. Define the 'Course' Entity Model:
class Course(Base):
__tablename__ = "courses"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
title: Mapped[str] = mapped_column(String(100), nullable=False)
instructor: Mapped[str] = mapped_column(String(50), nullable=False)
price: Mapped[float] = mapped_column(Float, default=0.0)
def __repr__(self) -> str:
return f""
# 3. Create the Database Engine (Connection Pool):
engine = create_engine("sqlite:///:memory:", echo=False)
# 4. Create all registered tables in database schema:
Base.metadata.create_all(engine)
# 5. Manage Transaction Lifecycle with a Session Context Manager:
with Session(engine) as session:
# CREATE:
c1 = Course(title="Python 3 Masterclass 2026", instructor="Balaji", price=1499.0)
c2 = Course(title="FastAPI Microservices", instructor="Alex", price=1999.0)
c3 = Course(title="Data Structures & Algorithms", instructor="Chloe", price=999.0)
session.add_all([c1, c2, c3])
session.commit()
print("✅ Created 3 Course records in database!")
# READ: Query with modern 2.0 select() statements:
query = select(Course).where(Course.price >= 1000.0).order_by(Course.price.desc())
premium_courses = session.scalars(query).all()
print("\n--- 🎓 Premium Courses (Price >= ₹1,000) ---")
for course in premium_courses:
print("•", course)
# UPDATE: Give discount to FastAPI course via dirty-checking:
fastapi_course = session.scalar(select(Course).where(Course.title.contains("FastAPI")))
if fastapi_course:
fastapi_course.price = 1299.0 # Just assign new attribute value!
session.commit() # SQLAlchemy auto-generates UPDATE query!
print(f"\n🏷️ Discount applied: {fastapi_course}")
🔍 SQLAlchemy 2.0 Architecture Breakdown:
Mapped[T]: PEP 484 type hint informing both Python type checkers and SQLAlchemy what SQL type to generate.session.scalars(query).all(): Returns a clean list ofCoursePython objects rather than raw database tuples.Dirty Checking: When you modifyfastapi_course.price, the Session detects the change and flushes the minimal SQL update uponcommit().
⚠️ Common Developer Pitfall: Mixing SQLAlchemy 1.x Query Syntax with 2.0 Code
In SQLAlchemy 2.0, legacy patterns like session.query(Course).filter(...) are deprecated. Always use the new 2.0 syntax: session.scalars(select(Course).where(...)).all().
💻 Hands-on Interactive Practice Challenge
Create an Author model in SQLAlchemy with id, name, and total_books. Create an in-memory session, add two authors, and select the author with the most books.
Python 3 Practice Challenge
▶ Run in Compiler
from sqlalchemy import create_engine, String, Integer, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
class Base(DeclarativeBase): pass
class Author(Base):
__tablename__ = "authors"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
name: Mapped[str] = mapped_column(String(50))
total_books: Mapped[int] = mapped_column(Integer)
engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)
with Session(engine) as session:
session.add_all([
Author(id=1, name="Luciano Ramalho", total_books=2),
Author(id=2, name="Robert C. Martin", total_books=6)
])
session.commit()
top = session.scalar(select(Author).order_by(Author.total_books.desc()))
print(f"Top Author: {top.name} with {top.total_books} books!")❓ Frequently Asked Questions (FAQ)
Q What is Alembic in the Python ecosystem?
Alembic is the official database migration tool for SQLAlchemy. It inspects your Python model classes and automatically generates versioned SQL upgrade/downgrade migration scripts as your database schema evolves.
Q What is Connection Pooling in SQLAlchemy?
Establishing a new TCP connection to PostgreSQL or MySQL takes 20-50 milliseconds. SQLAlchemy Engine maintains a pool of pre-opened active connections in memory, instantly reusing them for incoming requests and dramatically increasing web performance.
Q When should I write raw SQL instead of using an ORM?
For complex multi-table analytical reporting queries (with 8+ JOINs, window functions, and aggregations across millions of rows), raw SQL or SQLAlchemy Core (select() without ORM mapping) provides optimal query execution speed.