SQLAlchemy Core and ORM

Writing SQL by hand is fine for small scripts, but larger applications benefit from SQLAlchemy, the most popular Python database toolkit. It has two layers: Core, a Pythonic way to build and run SQL against any database, and the ORM (Object-Relational Mapper), which maps database tables to Python classes so you work with objects instead of rows.

The same code runs on SQLite, PostgreSQL, MySQL, SQL Server and Oracle — only the connection URL changes. This lesson uses the modern SQLAlchemy 2.0 style with type-annotated models. FastAPI, Flask and many other frameworks are commonly paired with SQLAlchemy.

SQLAlchemy is not available in the browser runner; install it with pip install sqlalchemy and run the examples locally. They use SQLite, so no server is needed.

Engine, Connection and Session

create_engine(url) creates an engine that manages a pool of connections. URLs look like sqlite:///app.db, postgresql+psycopg://user:pass@host/db or mysql+pymysql://user:pass@host/db. With Core you use engine.connect() / engine.begin() and text() or select(). With the ORM you use a Session: it tracks the objects you add or change and writes them to the database when you commit().

Declaring Models

Models subclass a DeclarativeBase. Each attribute is annotated with Mapped[type] and optionally configured with mapped_column() (primary keys, lengths, uniqueness, foreign keys). relationship() links models, for example one author to many books, with back_populates keeping both sides in sync. Base.metadata.create_all(engine) creates the tables; in real projects, use Alembic migrations to evolve the schema.

Querying

Build queries with select(Model).where(...).order_by(...).limit(...) and run them with session.scalars(query) (model objects) or session.execute(query) (rows). .one(), .first() and .all() fetch results; session.get(Model, id) loads by primary key. Joins and aggregates use .join(), .group_by() and func.count() / func.sum(). Values are always sent as bound parameters, so SQLAlchemy queries are safe from SQL injection.

Examples

SQLAlchemy Core: engine, text() and bound parameters

Python

# pip install sqlalchemy
from sqlalchemy import create_engine, text

engine = create_engine("sqlite:///:memory:")
# PostgreSQL: create_engine("postgresql+psycopg://user:password@localhost:5432/shop")
# MySQL:      create_engine("mysql+pymysql://user:password@localhost:3306/shop")

with engine.begin() as conn:                     # begin() commits at the end of the block
    conn.execute(text("CREATE TABLE stores (city TEXT, revenue INTEGER)"))
    conn.execute(text("INSERT INTO stores VALUES (:city, :revenue)"), [
        {"city": "Pune", "revenue": 820000},
        {"city": "Nashik", "revenue": 310000},
        {"city": "Nagpur", "revenue": 540000},
    ])

with engine.connect() as conn:
    result = conn.execute(text("SELECT city, revenue FROM stores WHERE revenue > :minimum ORDER BY revenue DESC"),
                          {"minimum": 400000})
    for row in result:
        print(row.city, row.revenue)

Output

Pune 820000
Nagpur 540000

ORM models, relationships and full CRUD

Python

from sqlalchemy import create_engine, select, func, String, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session

class Base(DeclarativeBase):
    pass

class Author(Base):
    __tablename__ = "authors"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100), unique=True)
    books: Mapped[list["Book"]] = relationship(back_populates="author", cascade="all, delete-orphan")

    def __repr__(self):
        return f"Author({self.name!r})"

class Book(Base):
    __tablename__ = "books"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    price: Mapped[int]
    author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"))
    author: Mapped[Author] = relationship(back_populates="books")

    def __repr__(self):
        return f"Book({self.title!r}, {self.price})"

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as session:
    # Create
    asha = Author(name="Asha", books=[Book(title="Python Basics", price=450),
                                      Book(title="Python Testing", price=599)])
    ravi = Author(name="Ravi", books=[Book(title="Deep Python", price=899)])
    session.add_all([asha, ravi])
    session.commit()
    print("ids:", asha.id, ravi.id)

    # Read
    book = session.scalars(select(Book).where(Book.title == "Deep Python")).one()
    print(book, "by", book.author.name)

    # Update: change the attribute, then commit
    book.price = 799
    session.commit()

    cheap = session.scalars(select(Book).where(Book.price < 700).order_by(Book.price)).all()
    print(cheap)

    stats = session.execute(
        select(Author.name, func.count(Book.id), func.sum(Book.price))
        .join(Author.books)
        .group_by(Author.name)
        .order_by(Author.name)
    ).all()
    print(stats)

    # Delete: the cascade removes Asha's books too
    session.delete(asha)
    session.commit()
    print(session.scalars(select(Book)).all())

Output

ids: 1 2
Book('Deep Python', 899) by Ravi
[Book('Python Basics', 450), Book('Python Testing', 599)]
[('Asha', 2, 1049), ('Ravi', 1, 799)]
[Book('Deep Python', 799)]

A session factory, get(), and rolling back on errors

Python

from sqlalchemy import create_engine, String
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)
SessionLocal = sessionmaker(bind=engine)          # create once, reuse everywhere

def register(email):
    with SessionLocal() as session:
        try:
            user = User(email=email)
            session.add(user)
            session.commit()
            return f"registered {email} as user {user.id}"
        except IntegrityError:
            session.rollback()
            return f"{email} is already registered"

print(register("asha@example.com"))
print(register("asha@example.com"))
with SessionLocal() as session:
    print(session.get(User, 1).email, session.get(User, 99))

Output

registered asha@example.com as user 1
asha@example.com is already registered
asha@example.com None

Schema migrations with Alembic

Python

pip install alembic
alembic init migrations                 # creates alembic.ini and migrations/
# set sqlalchemy.url in alembic.ini and target_metadata = Base.metadata in migrations/env.py
alembic revision --autogenerate -m "create authors and books"
alembic upgrade head                    # apply all migrations
alembic downgrade -1                    # undo the last one

Output

INFO  [alembic.runtime.migration] Running upgrade  -> 3f2a1c9d8e7b, create authors and books

Common Mistakes

Mixing the legacy 1.x query API (session.query) with 2.0-style select() without understanding the difference.
Forgetting session.commit(), so changes are lost when the session closes.
Triggering the "N+1 queries" problem by accessing relationships in a loop — use selectinload() or joinedload().
Using create_all() to change existing tables; it only creates missing tables — use Alembic migrations.
Sharing one Session across threads or web requests instead of one session per unit of work.

Key Points to Remember

SQLAlchemy Core builds and runs SQL; the ORM maps tables to classes.
create_engine(url) works with SQLite, PostgreSQL, MySQL and more — only the URL changes.
Declare models with DeclarativeBase, Mapped[...] and mapped_column(); link them with relationship().
Query with select(...) and session.scalars()/execute(); commit to save changes.
Use sessionmaker for sessions and Alembic for schema migrations.

Practice the examples

Change an input, predict the result, then compare it with the output. Explain why the result changes.

Use your local project environment for these examples. Codelab currently runs Python and HTML/CSS/JavaScript; framework examples may need project dependencies.