SQLAlchemy - ORM dla Python

Witaj! Ostatnio pisałeś/aś surowe SQL - cursor.execute("SELECT * FROM species"). Działa, ale każde zapytanie to napis, a literówka w nazwie kolumny wychodzi dopiero przy uruchomieniu. Poznaj ORM (Object-Relational Mapping) - pracę z bazą przez pythonowe klasy zamiast SQL!

Analogia Safari: Surowe SQL to jak wypełnianie formularzy papierowych ręcznie. SQLAlchemy ORM to cyfrowy system - obiekty Python automatycznie mapowane na tabele bazy danych!

Czym jest ORM?

ORM mapuje obiekty na tabele bazy: klasa to tabela, obiekt to wiersz, atrybut to kolumna. W surowym SQL zapis lwa to napis ze znakami zapytania zamiast wartości:

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

W ORM tworzysz zwykły obiekt Pythona, a gotowe polecenie INSERT przygotuje za Ciebie sesja:

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

Oba dodają ten sam wiersz, ale w drugim SQL pisze SQLAlchemy.

Zalety ORM:

• Pythonowe klasy zamiast SQL stringów
• Type hints i autocomplete
• Mniej podatne na SQL injection
• Łatwiejsze relacje między tabelami
• Migracje i ewolucja schematu
• Niezależność od bazy (PostgreSQL, MySQL, SQLite...)

SQLAlchemy - najpopularniejszy ORM w Pythonie

SQLAlchemy to potężny ORM + narzędzie do pracy z bazami SQL. Używamy aktualnej serii 2.0.

Instalacja

Nie ma jej w standardowym Pythonie, więc instalujemy ją raz z terminala:

1pip install sqlalchemy

Dla SQLite to wszystko - sterownik sqlite3 jest wbudowany w Pythona.

Podstawy SQLAlchemy - Modele

Model to klasa opisująca tabelę: nazwa trafia do __tablename__, a kolumny to atrybuty Column z typami i ograniczeniami. Pięć kroków prowadzi do gotowej sesji:

1from sqlalchemy import create_engine, Column, Integer, String, Boolean
2from sqlalchemy.orm import declarative_base, sessionmaker
3
4# 1. Utwórz Base - klasa bazowa dla modeli
5Base = declarative_base()
6
7# 2. Zdefiniuj model - klasa Python = tabela SQL
8class Species(Base):
9    __tablename__ = 'species'  # Nazwa tabeli
10
11    # Kolumny
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. Połącz się z bazą danych
24engine = create_engine('sqlite:///safari_orm.db', echo=True)  # echo=True wyświetla SQL
25
26# 4. Utwórz tabele
27Base.metadata.create_all(engine)
28
29# 5. Utwórz sesję (zarządza operacjami na bazie)
30Session = sessionmaker(bind=engine)
31session = Session()

primary_key=True to unikalny numer wiersza, nullable=False zakazuje pustych wartości, unique=True duplikatów. Podczas nauki włącz echo=True, by widzieć generowany SQL. Uwaga: Base.metadata.create_all() tworzy tabele tylko jeśli nie istnieją (jak IF NOT EXISTS).

Typy kolumn

MySQL wymaga długości tekstu, np. String(100).

CRUD z SQLAlchemy

CREATE - dodawanie rekordów

Nowy rekord to nowy obiekt modelu. Trafia do sesji, a do bazy dopiero przy zatwierdzeniu:

1# Utwórz obiekt
2lion = Species(
3    scientific_name="Panthera leo",
4    common_name="Lew",
5    population=120,
6    habitat="sawanna",
7    endangered=True
8)
9
10# Dodaj do sesji
11session.add(lion)
12
13# Zatwierdź
14session.commit()
15
16print(f"Dodano lwa ID: {lion.id}")  # ID automatycznie przypisane!
17
18# Dodaj wiele naraz
19species_list = [
20    Species(scientific_name="Loxodonta africana", common_name="Słoń", population=450, endangered=True),
21    Species(scientific_name="Gorilla gorilla", common_name="Goryl", population=230, endangered=True),
22]
23
24session.add_all(species_list)
25session.commit()

Przed commit() lew żyje tylko w sesji, a jego id to None.

READ - pobieranie danych

Zapytanie zaczyna się od session.query(Species), a metody doklejasz jak ogniwa łańcucha, kończąc tą, która pobiera wyniki:

1# Wszystkie gatunki
2all_species = session.query(Species).all()
3for species in all_species:
4    print(species.common_name, species.population)
5
6# Pierwszy wynik
7first = session.query(Species).first()
8
9# Pobierz po ID
10lion = session.query(Species).get(1)  # ID = 1 (styl legacy - ostrzeżenie w 2.0)
11# lub (SQLAlchemy 1.4+, zalecane)
12lion = session.get(Species, 1)
13
14# Filtrowanie - WHERE
15endangered = session.query(Species).filter(Species.endangered == True).all()
16savanna = session.query(Species).filter(Species.habitat == "sawanna").all()
17
18# Wiele warunków (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 == "sawanna", Species.habitat == "las")
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 i 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() przyjmuje operatory Pythona, a SQLAlchemy zamienia je na WHERE. Styl session.query() ma w 2.0 etykietę "legacy" - nowszy select() poznasz w module o FastAPI.

UPDATE - aktualizacja

Aktualizacja w ORM to po prostu zmiana atrybutu. Sesja ją zauważa i przy commit() wysyła polecenie UPDATE:

1# Metoda 1: Pobierz obiekt, zmień, commit
2lion = session.query(Species).filter(Species.common_name == "Lew").first()
3lion.population = 125
4session.commit()
5
6# Metoda 2: Bulk update
7session.query(Species).filter(Species.habitat == "sawanna").update({
8    "endangered": True
9})
10session.commit()

Metoda 1 wczytuje obiekt, metoda 2 zmienia wiele wierszy jednym poleceniem bez ich ładowania.

DELETE - usuwanie

Pojedynczy obiekt przekazujesz do session.delete(), a wiele wierszy usuwa filtr zakończony metodą delete():

1# Metoda 1: Pobierz obiekt, usuń
2species_to_delete = session.get(Species, 10)
3if species_to_delete:
4    session.delete(species_to_delete)
5    session.commit()
6
7# Metoda 2: Bulk delete
8session.query(Species).filter(Species.population == 0).delete()
9session.commit()

session.get() zwraca None, gdy rekordu nie ma, stąd if. Bez commit() zmiana przepadnie.

Relacje - One-to-Many

Jeden gatunek ma wiele obserwacji z terenu: w bazie łączy je klucz obcy, a w Pythonie relationship() (osobny, uproszczony przykład):

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    # Relacja: jeden gatunek -> wiele obserwacji
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    # Relacja: wiele obserwacji -> jeden gatunek
25    species = relationship("Species", back_populates="observations")

ForeignKey('species.id') wskazuje kolumnę gatunków, back_populates spina oba końce, a ustawienie cascade usuwa obserwacje razem z gatunkiem.

Teraz dopisujemy obserwacje lwa do jego listy i zapisujemy wszystko w bazie jednym commit():

1# Użycie
2lion = Species(common_name="Lew", population=120)
3
4# Dodaj obserwacje do lwa
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# Pobierz obserwacje
12lion = session.query(Species).filter(Species.common_name == "Lew").first()
13for obs in lion.observations:
14    print(f"{obs.observation_date}: {obs.count}x w {obs.location}")

Pola species_id nie ustawialiśmy - SQLAlchemy uzupełnił je sam dzięki relacji.

Safari przykład - kompletny system ORM

Zbierzmy wszystko w klasę-usługę, która ukrywa sesję. Najpierw modele ze znacznikami czasu:

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    """Model gatunku"""
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    # Relacje
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    """Model obserwacji"""
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    # Relacje
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 i onupdate dostają funkcję utc_now, a nie jej wynik, więc czas liczy się przy każdym zapisie. datetime.utcnow() jest od Pythona 3.12 przestarzałe.

Klasa SafariORM tworzy silnik i sesję w konstruktorze, a każda jej metoda wykonuje jedną operację CRUD:

1class SafariORM:
2    """Zarządzanie bazą danych Safari przez 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        """Dodaj gatunek"""
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        """Pobierz gatunek"""
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        """Lista gatunków"""
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        """Zaktualizuj gatunek"""
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        """Usuń gatunek"""
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        """Dodaj obserwację"""
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        """Obserwacje gatunku"""
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        """Zamknij sesję"""
95        self.session.close()

Metody zwracają modele albo None, więc reszta programu nie dotyka SQL ani sesji.

Demonstracja używa bazy w pamięci, więc kolejne uruchomienia nie kłócą się o unikalne nazwy:

1# === DEMONSTRACJA ===
2
3print("=== SAFARI ORM SYSTEM ===\n")
4
5db = SafariORM("sqlite:///:memory:")
6
7# 1. Dodaj gatunki
8print("1. Dodawanie gatunków (ORM)...")
9lion = db.create_species("Panthera leo", "Lew", 120, "sawanna", True)
10elephant = db.create_species("Loxodonta africana", "Słoń", 450, "sawanna", True)
11gorilla = db.create_species("Gorilla gorilla", "Goryl", 230, "las", True)
12
13print(f"   Dodano: {lion}, {elephant}, {gorilla}")
14
15# 2. Pobierz gatunek
16print("\n2. Pobieranie gatunku...")
17retrieved_lion = db.get_species(lion.id)
18print(f"   {retrieved_lion.common_name}: {retrieved_lion.population} osobników")
19
20# 3. Lista zagrożonych
21print("\n3. Lista zagrożonych gatunków...")
22endangered = db.list_species(endangered=True)
23for species in endangered:
24    print(f"   - {species.common_name}: {species.population}")
25
26# 4. Aktualizacja
27print("\n4. Aktualizacja populacji...")
28db.update_species(lion.id, population=125)
29lion = db.get_species(lion.id)
30print(f"   Nowa populacja lwa: {lion.population}")
31
32# 5. Dodaj obserwacje
33print("\n5. Dodawanie obserwacji...")
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. Pobierz obserwacje
38print("\n6. Obserwacje lwa...")
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. Relacje
44print("\n7. Nawigacja przez relacje...")
45lion = db.get_species(lion.id)
46print(f"   Lew ma {len(lion.observations)} obserwacji:")
47for obs in lion.observations:
48    print(f"     {obs.location}: {obs.count}x")
49
50db.close()
51print("\nDemonstracja zakończona")

Adres bazy to zawsze URL, np. sqlite:///plik.db - sama nazwa pliku skończy się błędem ArgumentError.

Zajrzyjmy do następnego obozu

Za chwilę Darwin pokaże Ci NoSQL MongoDB - bazę dokumentową bez sztywnego schematu. Te same kroki wyglądają tam tak (wymaga serwera MongoDB):

1from pymongo import MongoClient
2
3client = MongoClient()      # połączenie z serwerem MongoDB
4db = client["safari"]       # wybór bazy
5db.animals.insert_one({"name": "Lew", "population": 120})
6results = db.animals.find() # kursor z dokumentami

Nie ma tu modelu ani create_all() - kolekcja animals powstaje przy pierwszym zapisie.

Podsumowanie

W tej lekcji nauczyłeś/aś się:

• Czym jest ORM i po co służy
• SQLAlchemy: Base, modele, kolumny
• CRUD z ORM: add, query, update, delete
• Filtrowanie: filter(), order_by(), limit()
• Relacje: One-to-Many, relationship, ForeignKey
• Kompletny system Safari z ORM

Pamiętaj: ORM to cyfrowy katalog - oddajesz mu pythonowe obiekty, a formularze SQL wypełniają się za Ciebie.