SQLAlchemy - ORM for Python

Welcome! Last lesson you wrote raw SQL - cursor.execute("SELECT * FROM species"). It works, but every query is a string, and a typo in a column name shows up only at runtime. Now meet ORM (Object-Relational Mapping) - working with a database through Python classes instead of SQL!

Safari Analogy: Raw SQL is like filling out paper forms by hand. SQLAlchemy ORM is a digital system - Python objects automatically mapped to database tables!

What is ORM?

ORM maps objects to database tables: a class is a table, an object a row, an attribute a column. In raw SQL, saving a lion is a string with question marks for the values:

1cursor.execute("INSERT INTO species (name, population) VALUES (?, ?)", ("Lion", 120))

With an ORM you create a regular object, and the session prepares the INSERT itself:

1lion = Species(name="Lion", population=120)
2session.add(lion)
3session.commit()

Both add the same row, but in the second SQLAlchemy writes the SQL.

ORM advantages:

• Python classes instead of SQL strings
• Type hints and autocomplete
• Less vulnerable to SQL injection
• Easier relationships between tables
• Migrations and schema evolution
• Database independence (PostgreSQL, MySQL, SQLite...)

SQLAlchemy - the most popular ORM in Python

SQLAlchemy is a powerful ORM + tool for working with SQL databases. We use the current 2.0 series.

Installation

It is not part of standard Python, so install it once from the terminal:

1pip install sqlalchemy

For SQLite that's all - the sqlite3 driver is built into Python.

SQLAlchemy Basics - Models

A model is a class describing a table: its name goes in __tablename__, its columns are Column attributes with types and constraints. Five steps lead to a ready session:

1from sqlalchemy import create_engine, Column, Integer, String, Boolean
2from sqlalchemy.orm import declarative_base, sessionmaker
3
4# 1. Create Base - base class for models
5Base = declarative_base()
6
7# 2. Define model - Python class = SQL table
8class Species(Base):
9    __tablename__ = 'species'  # Table name
10
11    # Columns
12    id = Column(Integer, primary_key=True, autoincrement=True)
13    scientific_name = Column(String, nullable=False, unique=True)
14    common_name = Column(String, nullable=False)
15    population = Column(Integer, default=0)
16    habitat = Column(String)
17    endangered = Column(Boolean, default=False)
18
19    def __repr__(self):
20        return f"<Species('{self.common_name}', pop={self.population})>"
21
22
23# 3. Connect to database
24engine = create_engine('sqlite:///safari_orm.db', echo=True)  # echo=True displays SQL
25
26# 4. Create tables
27Base.metadata.create_all(engine)
28
29# 5. Create session (manages database operations)
30Session = sessionmaker(bind=engine)
31session = Session()

primary_key=True is the unique row number, nullable=False forbids empty values, unique=True duplicates. Use echo=True while learning to see the generated SQL. Note: Base.metadata.create_all() creates tables only if they don't exist (like IF NOT EXISTS).

Column Types

MySQL requires a text length, e.g. String(100).

CRUD with SQLAlchemy

CREATE - adding records

A new record is a new model object. It goes into the session and reaches the database only on commit:

1# Create object
2lion = Species(
3    scientific_name="Panthera leo",
4    common_name="Lion",
5    population=120,
6    habitat="savanna",
7    endangered=True
8)
9
10# Add to session
11session.add(lion)
12
13# Commit
14session.commit()
15
16print(f"Added lion ID: {lion.id}")  # ID automatically assigned!
17
18# Add multiple at once
19species_list = [
20    Species(scientific_name="Loxodonta africana", common_name="Elephant", population=450, endangered=True),
21    Species(scientific_name="Gorilla gorilla", common_name="Gorilla", population=230, endangered=True),
22]
23
24session.add_all(species_list)
25session.commit()

Before commit() the lion lives only in the session, with id equal to None.

READ - retrieving data

A query starts with session.query(Species), and you chain methods like links, ending with the one that fetches results:

1# All species
2all_species = session.query(Species).all()
3for species in all_species:
4    print(species.common_name, species.population)
5
6# First result
7first = session.query(Species).first()
8
9# Get by ID
10lion = session.query(Species).get(1)  # ID = 1 (legacy style - warning in 2.0)
11# or (SQLAlchemy 1.4+, recommended)
12lion = session.get(Species, 1)
13
14# Filtering - WHERE
15endangered = session.query(Species).filter(Species.endangered == True).all()
16savanna = session.query(Species).filter(Species.habitat == "savanna").all()
17
18# Multiple conditions (AND)
19results = session.query(Species).filter(
20    Species.endangered == True,
21    Species.population > 100
22).all()
23
24# OR
25from sqlalchemy import or_
26results = session.query(Species).filter(
27    or_(Species.habitat == "savanna", Species.habitat == "forest")
28).all()
29
30# LIKE
31results = session.query(Species).filter(Species.scientific_name.like("Panthera%")).all()
32
33# ORDER BY
34sorted_species = session.query(Species).order_by(Species.population.desc()).all()
35
36# LIMIT and OFFSET
37top_5 = session.query(Species).limit(5).all()
38page_2 = session.query(Species).limit(10).offset(10).all()
39
40# COUNT
41count = session.query(Species).count()
42endangered_count = session.query(Species).filter(Species.endangered == True).count()

filter() takes Python operators and SQLAlchemy turns them into WHERE. The session.query() style is "legacy" in 2.0 - the newer select() comes in the FastAPI module.

UPDATE - updating

An ORM update is just an attribute change. The session notices it and sends an UPDATE statement on commit():

1# Method 1: Fetch object, modify, commit
2lion = session.query(Species).filter(Species.common_name == "Lion").first()
3lion.population = 125
4session.commit()
5
6# Method 2: Bulk update
7session.query(Species).filter(Species.habitat == "savanna").update({
8    "endangered": True
9})
10session.commit()

Method 1 loads the object, method 2 changes many rows in one statement without loading them.

DELETE - deleting

You pass a single object to session.delete(), and many rows are removed by a filter ending with delete():

1# Method 1: Fetch object, delete
2species_to_delete = session.get(Species, 10)
3if species_to_delete:
4    session.delete(species_to_delete)
5    session.commit()
6
7# Method 2: Bulk delete
8session.query(Species).filter(Species.population == 0).delete()
9session.commit()

session.get() returns None when the record doesn't exist, hence the if. Without commit() the change is lost.

Relationships - One-to-Many

One species has many field observations: a foreign key links them in the database, relationship() in Python (a separate, simplified example):

1from sqlalchemy import ForeignKey
2from sqlalchemy.orm import relationship
3
4class Species(Base):
5    __tablename__ = 'species'
6
7    id = Column(Integer, primary_key=True)
8    common_name = Column(String, nullable=False)
9    population = Column(Integer, default=0)
10
11    # Relationship: one species -> many observations
12    observations = relationship("Observation", back_populates="species", cascade="all, delete-orphan")
13
14
15class Observation(Base):
16    __tablename__ = 'observations'
17
18    id = Column(Integer, primary_key=True)
19    species_id = Column(Integer, ForeignKey('species.id'), nullable=False)
20    observation_date = Column(String, nullable=False)
21    location = Column(String, nullable=False)
22    count = Column(Integer, default=0)
23
24    # Relationship: many observations -> one species
25    species = relationship("Species", back_populates="observations")

ForeignKey('species.id') points to the species column, back_populates connects both ends, and the cascade setting deletes observations with their species.

Now we append the lion's observations to its list and save everything to the database with one commit():

1# Usage
2lion = Species(common_name="Lion", population=120)
3
4# Add observations to the lion
5lion.observations.append(Observation(observation_date="2024-01-15", location="Serengeti", count=12))
6lion.observations.append(Observation(observation_date="2024-01-20", location="Masai Mara", count=8))
7
8session.add(lion)
9session.commit()
10
11# Fetch observations
12lion = session.query(Species).filter(Species.common_name == "Lion").first()
13for obs in lion.observations:
14    print(f"{obs.observation_date}: {obs.count}x in {obs.location}")

We never set species_id - SQLAlchemy filled it in thanks to the relationship.

Safari example - complete ORM system

Let's gather everything into a service class that hides the session. First, the models with timestamps:

1from sqlalchemy import create_engine, Column, Integer, String, Boolean, ForeignKey, DateTime
2from sqlalchemy.orm import declarative_base, sessionmaker, relationship
3from datetime import datetime, timezone
4from typing import List, Optional
5
6Base = declarative_base()
7
8
9def utc_now() -> datetime:
10    return datetime.now(timezone.utc)
11
12
13class Species(Base):
14    """Species model"""
15    __tablename__ = 'species'
16
17    id = Column(Integer, primary_key=True, autoincrement=True)
18    scientific_name = Column(String, nullable=False, unique=True)
19    common_name = Column(String, nullable=False)
20    population = Column(Integer, default=0)
21    habitat = Column(String)
22    endangered = Column(Boolean, default=False)
23    created_at = Column(DateTime, default=utc_now)
24    updated_at = Column(DateTime, default=utc_now, onupdate=utc_now)
25
26    # Relationships
27    observations = relationship("Observation", back_populates="species", cascade="all, delete-orphan")
28
29    def __repr__(self):
30        return f"<Species('{self.common_name}', pop={self.population})>"
31
32
33class Observation(Base):
34    """Observation model"""
35    __tablename__ = 'observations'
36
37    id = Column(Integer, primary_key=True, autoincrement=True)
38    species_id = Column(Integer, ForeignKey('species.id'), nullable=False)
39    observation_date = Column(String, nullable=False)
40    location = Column(String, nullable=False)
41    count = Column(Integer, default=0)
42    notes = Column(String)
43    created_at = Column(DateTime, default=utc_now)
44
45    # Relationships
46    species = relationship("Species", back_populates="observations")
47
48    def __repr__(self):
49        return f"<Observation({self.species.common_name if self.species else 'N/A'}, {self.count}x @ {self.location})>"

default and onupdate get the utc_now function, not its result, so the time is computed on every write. datetime.utcnow() is deprecated since Python 3.12.

The SafariORM class creates the engine and session in its constructor, and each of its methods performs one CRUD operation:

1class SafariORM:
2    """Safari database management via ORM"""
3
4    def __init__(self, db_url: str = "sqlite:///safari_orm.db"):
5        self.engine = create_engine(db_url, echo=False)
6        Base.metadata.create_all(self.engine)
7        Session = sessionmaker(bind=self.engine)
8        self.session = Session()
9
10    # === SPECIES ===
11
12    def create_species(self, scientific_name: str, common_name: str,
13                      population: int = 0, habitat: str = "",
14                      endangered: bool = False) -> Species:
15        """Add a species"""
16        species = Species(
17            scientific_name=scientific_name,
18            common_name=common_name,
19            population=population,
20            habitat=habitat,
21            endangered=endangered
22        )
23        self.session.add(species)
24        self.session.commit()
25        return species
26
27    def get_species(self, species_id: int) -> Optional[Species]:
28        """Get a species"""
29        return self.session.get(Species, species_id)
30
31    def list_species(self, endangered: Optional[bool] = None,
32                    habitat: Optional[str] = None) -> List[Species]:
33        """List species"""
34        query = self.session.query(Species)
35
36        if endangered is not None:
37            query = query.filter(Species.endangered == endangered)
38        if habitat:
39            query = query.filter(Species.habitat == habitat)
40
41        return query.order_by(Species.common_name).all()
42
43    def update_species(self, species_id: int, **kwargs) -> Optional[Species]:
44        """Update a species"""
45        species = self.session.get(Species, species_id)
46        if not species:
47            return None
48
49        for key, value in kwargs.items():
50            if hasattr(species, key):
51                setattr(species, key, value)
52
53        species.updated_at = utc_now()
54        self.session.commit()
55        return species
56
57    def delete_species(self, species_id: int) -> bool:
58        """Delete a species"""
59        species = self.session.get(Species, species_id)
60        if not species:
61            return False
62
63        self.session.delete(species)
64        self.session.commit()
65        return True
66
67    # === OBSERVATIONS ===
68
69    def create_observation(self, species_id: int, observation_date: str,
70                          location: str, count: int, notes: str = "") -> Optional[Observation]:
71        """Add an observation"""
72        species = self.session.get(Species, species_id)
73        if not species:
74            return None
75
76        observation = Observation(
77            species_id=species_id,
78            observation_date=observation_date,
79            location=location,
80            count=count,
81            notes=notes
82        )
83        self.session.add(observation)
84        self.session.commit()
85        return observation
86
87    def get_observations_for_species(self, species_id: int) -> List[Observation]:
88        """Get observations for a species"""
89        return self.session.query(Observation).filter(
90            Observation.species_id == species_id
91        ).order_by(Observation.observation_date.desc()).all()
92
93    def close(self):
94        """Close session"""
95        self.session.close()

Methods return models or None, so the rest of the program never touches SQL or the session.

The demo uses an in-memory database, so repeated runs never clash on unique names:

1# === DEMONSTRATION ===
2
3print("=== SAFARI ORM SYSTEM ===\n")
4
5db = SafariORM("sqlite:///:memory:")
6
7# 1. Add species
8print("1. Adding species (ORM)...")
9lion = db.create_species("Panthera leo", "Lion", 120, "savanna", True)
10elephant = db.create_species("Loxodonta africana", "Elephant", 450, "savanna", True)
11gorilla = db.create_species("Gorilla gorilla", "Gorilla", 230, "forest", True)
12
13print(f"   Added: {lion}, {elephant}, {gorilla}")
14
15# 2. Fetch species
16print("\n2. Fetching species...")
17retrieved_lion = db.get_species(lion.id)
18print(f"   {retrieved_lion.common_name}: {retrieved_lion.population} individuals")
19
20# 3. List endangered
21print("\n3. List of endangered species...")
22endangered = db.list_species(endangered=True)
23for species in endangered:
24    print(f"   - {species.common_name}: {species.population}")
25
26# 4. Update
27print("\n4. Updating population...")
28db.update_species(lion.id, population=125)
29lion = db.get_species(lion.id)
30print(f"   New lion population: {lion.population}")
31
32# 5. Add observations
33print("\n5. Adding observations...")
34db.create_observation(lion.id, "2024-01-15", "Serengeti", 12, "Pride with cubs")
35db.create_observation(lion.id, "2024-01-20", "Masai Mara", 8, "Male coalition")
36
37# 6. Fetch observations
38print("\n6. Lion observations...")
39observations = db.get_observations_for_species(lion.id)
40for obs in observations:
41    print(f"   - {obs.observation_date}: {obs.count}x @ {obs.location}")
42
43# 7. Relationships
44print("\n7. Navigating through relationships...")
45lion = db.get_species(lion.id)
46print(f"   Lion has {len(lion.observations)} observations:")
47for obs in lion.observations:
48    print(f"     {obs.location}: {obs.count}x")
49
50db.close()
51print("\nDemonstration complete")

A database address is always a URL such as sqlite:///file.db - a bare file name raises an ArgumentError.

A Look at the Next Camp

Next, Darwin shows you NoSQL MongoDB - a document database without a rigid schema. The same steps look like this (a MongoDB server is required):

1from pymongo import MongoClient
2
3client = MongoClient()      # connection to the MongoDB server
4db = client["safari"]       # choose a database
5db.animals.insert_one({"name": "Lion", "population": 120})
6results = db.animals.find() # cursor with documents

There is no model and no create_all() - the animals collection appears on the first write.

Summary

In this lesson you learned:

• What ORM is and what it's for
• SQLAlchemy: Base, models, columns
• CRUD with ORM: add, query, update, delete
• Filtering: filter(), order_by(), limit()
• Relationships: One-to-Many, relationship, ForeignKey
• A complete Safari system with ORM

Remember: an ORM is a digital catalog - you hand over Python objects, and the SQL forms get filled in for you.