SQLAlchemy - ORM for Python

Welcome, @name! Darwin here with a fascinating lesson about ORM.

In the previous lesson you learned raw SQL -

cursor.execute("SELECT * FROM species")

. It works great, but... writing SQL as strings is inconvenient, error-prone, and doesn't leverage Python classes!

Now you'll learn ORM (Object-Relational Mapping) - a way to work with databases using 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 (Object-Relational Mapping) is a technique for mapping programming objects to tables in a relational database.

Instead of:

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

You write:

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

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.

Installation

1pip install sqlalchemy

SQLAlchemy Basics - Models

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()

Note:

Base.metadata.create_all()

creates tables only if they don't exist (like

IF NOT EXISTS

)

Column Types

| SQLAlchemy | SQL | Python | |------------|-----|--------| |

Integer

| INTEGER |

int

| |

String

| VARCHAR |

str

| |

Text

| TEXT |

str

(longer) | |

Float

| REAL |

float

| |

Boolean

| BOOLEAN |

bool

| |

DateTime

| TIMESTAMP |

datetime

| |

Date

| DATE |

date

| |

JSON

| JSON |

dict

/

list

(SQLite 3.9+) |

CRUD with SQLAlchemy

CREATE - adding records

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

READ - retrieving data

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
11# or (SQLAlchemy 2.0+)
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()

UPDATE - updating

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()

DELETE - deleting

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

Relationships - One-to-Many

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")
26
27
28# Usage
29lion = Species(common_name="Lion", population=120)
30
31# Add observations to the lion
32lion.observations.append(Observation(observation_date="2024-01-15", location="Serengeti", count=12))
33lion.observations.append(Observation(observation_date="2024-01-20", location="Masai Mara", count=8))
34
35session.add(lion)
36session.commit()
37
38# Fetch observations
39lion = session.query(Species).filter(Species.common_name == "Lion").first()
40for obs in lion.observations:
41    print(f"{obs.observation_date}: {obs.count}x in {obs.location}")

Safari example - complete ORM system

1from sqlalchemy import create_engine, Column, Integer, String, Boolean, ForeignKey, DateTime
2from sqlalchemy.orm import declarative_base, sessionmaker, relationship
3from datetime import datetime
4from typing import List, Optional
5
6Base = declarative_base()
7
8
9class Species(Base):
10    """Species model"""
11    __tablename__ = 'species'
12
13    id = Column(Integer, primary_key=True, autoincrement=True)
14    scientific_name = Column(String, nullable=False, unique=True)
15    common_name = Column(String, nullable=False)
16    population = Column(Integer, default=0)
17    habitat = Column(String)
18    endangered = Column(Boolean, default=False)
19    created_at = Column(DateTime, default=datetime.utcnow)
20    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
21
22    # Relationships
23    observations = relationship("Observation", back_populates="species", cascade="all, delete-orphan")
24
25    def __repr__(self):
26        return f"<Species('{self.common_name}', pop={self.population})>"
27
28
29class Observation(Base):
30    """Observation model"""
31    __tablename__ = 'observations'
32
33    id = Column(Integer, primary_key=True, autoincrement=True)
34    species_id = Column(Integer, ForeignKey('species.id'), nullable=False)
35    observation_date = Column(String, nullable=False)
36    location = Column(String, nullable=False)
37    count = Column(Integer, default=0)
38    notes = Column(String)
39    created_at = Column(DateTime, default=datetime.utcnow)
40
41    # Relationships
42    species = relationship("Species", back_populates="observations")
43
44    def __repr__(self):
45        return f"<Observation({self.species.common_name if self.species else 'N/A'}, {self.count}x @ {self.location})>"
46
47
48class SafariORM:
49    """Safari database management via ORM"""
50
51    def __init__(self, db_url: str = "sqlite:///safari_orm.db"):
52        self.engine = create_engine(db_url, echo=False)
53        Base.metadata.create_all(self.engine)
54        Session = sessionmaker(bind=self.engine)
55        self.session = Session()
56
57    # === SPECIES ===
58
59    def create_species(self, scientific_name: str, common_name: str,
60                      population: int = 0, habitat: str = "",
61                      endangered: bool = False) -> Species:
62        """Add a species"""
63        species = Species(
64            scientific_name=scientific_name,
65            common_name=common_name,
66            population=population,
67            habitat=habitat,
68            endangered=endangered
69        )
70        self.session.add(species)
71        self.session.commit()
72        return species
73
74    def get_species(self, species_id: int) -> Optional[Species]:
75        """Get a species"""
76        return self.session.get(Species, species_id)
77
78    def list_species(self, endangered: Optional[bool] = None,
79                    habitat: Optional[str] = None) -> List[Species]:
80        """List species"""
81        query = self.session.query(Species)
82
83        if endangered is not None:
84            query = query.filter(Species.endangered == endangered)
85        if habitat:
86            query = query.filter(Species.habitat == habitat)
87
88        return query.order_by(Species.common_name).all()
89
90    def update_species(self, species_id: int, **kwargs) -> Optional[Species]:
91        """Update a species"""
92        species = self.session.get(Species, species_id)
93        if not species:
94            return None
95
96        for key, value in kwargs.items():
97            if hasattr(species, key):
98                setattr(species, key, value)
99
100        species.updated_at = datetime.utcnow()
101        self.session.commit()
102        return species
103
104    def delete_species(self, species_id: int) -> bool:
105        """Delete a species"""
106        species = self.session.get(Species, species_id)
107        if not species:
108            return False
109
110        self.session.delete(species)
111        self.session.commit()
112        return True
113
114    # === OBSERVATIONS ===
115
116    def create_observation(self, species_id: int, observation_date: str,
117                          location: str, count: int, notes: str = "") -> Optional[Observation]:
118        """Add an observation"""
119        species = self.session.get(Species, species_id)
120        if not species:
121            return None
122
123        observation = Observation(
124            species_id=species_id,
125            observation_date=observation_date,
126            location=location,
127            count=count,
128            notes=notes
129        )
130        self.session.add(observation)
131        self.session.commit()
132        return observation
133
134    def get_observations_for_species(self, species_id: int) -> List[Observation]:
135        """Get observations for a species"""
136        return self.session.query(Observation).filter(
137            Observation.species_id == species_id
138        ).order_by(Observation.observation_date.desc()).all()
139
140    def close(self):
141        """Close session"""
142        self.session.close()
143
144
145# === DEMONSTRATION ===
146
147print("=== SAFARI ORM SYSTEM ===\n")
148
149db = SafariORM("safari_orm_demo.db")
150
151# 1. Add species
152print("1. Adding species (ORM)...")
153lion = db.create_species("Panthera leo", "Lion", 120, "savanna", True)
154elephant = db.create_species("Loxodonta africana", "Elephant", 450, "savanna", True)
155gorilla = db.create_species("Gorilla gorilla", "Gorilla", 230, "forest", True)
156
157print(f"   Added: {lion}, {elephant}, {gorilla}")
158
159# 2. Fetch species
160print("\n2. Fetching species...")
161retrieved_lion = db.get_species(lion.id)
162print(f"   {retrieved_lion.common_name}: {retrieved_lion.population} individuals")
163
164# 3. List endangered
165print("\n3. List of endangered species...")
166endangered = db.list_species(endangered=True)
167for species in endangered:
168    print(f"   - {species.common_name}: {species.population}")
169
170# 4. Update
171print("\n4. Updating population...")
172db.update_species(lion.id, population=125)
173lion = db.get_species(lion.id)
174print(f"   New lion population: {lion.population}")
175
176# 5. Add observations
177print("\n5. Adding observations...")
178db.create_observation(lion.id, "2024-01-15", "Serengeti", 12, "Pride with cubs")
179db.create_observation(lion.id, "2024-01-20", "Masai Mara", 8, "Male coalition")
180
181# 6. Fetch observations
182print("\n6. Lion observations...")
183observations = db.get_observations_for_species(lion.id)
184for obs in observations:
185    print(f"   - {obs.observation_date}: {obs.count}x @ {obs.location}")
186
187# 7. Relationships
188print("\n7. Navigating through relationships...")
189lion = db.get_species(lion.id)
190print(f"   Lion has {len(lion.observations)} observations:")
191for obs in lion.observations:
192    print(f"     {obs.location}: {obs.count}x")
193
194db.close()
195print("\nDemonstration complete")

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

Safari Analogy: ORM is a digital catalog system - instead of manually filling out SQL forms, you use Python objects that automatically map to the database!

In the next (and last!) lesson of this module, Darwin will show you NoSQL MongoDB - a document database without a rigid schema!