Praca z bazą danych w automatyzacji: SQLite/PostgreSQL

Praca z SQLite

SQLite to lekkie, wbudowane rozwiązanie bazodanowe, które nie wymaga osobnego serwera i przechowuje całą bazę w jednym pliku. To idealny wybór do lokalnych projektów, prototypów, skryptów automatyzacyjnych oraz sytuacji, gdy nie potrzebujesz zaawansowanych funkcji relacyjnych baz danych. Python posiada wbudowaną obsługę SQLite poprzez moduł sqlite3, a użycie context managera gwarantuje poprawne zarządzanie połączeniami i transakcjami. Oto przykład efektywnej pracy z SQLite:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import sqlite3
from contextlib import contextmanager

@contextmanager
def get_db_connection(db_path: str):
    """Context manager dla połączenia z bazą SQLite."""
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

# Przykład użycia SQLite
with get_db_connection("dane.db") as conn:
    cursor = conn.cursor()
    cursor.execute("CREATE TABLE IF NOT EXISTS użytkownicy (id INTEGER PRIMARY KEY, imię TEXT, email TEXT)")
    cursor.execute("INSERT INTO użytkownicy (imię, email) VALUES (?, ?)", ("Jan", "jan@example.com"))

    # Odczyt danych
    cursor.execute("SELECT * FROM użytkownicy")
    for row in cursor.fetchall():
        print(dict(row))

Praca z PostgreSQL

PostgreSQL to zaawansowany system zarządzania bazą danych, który oferuje pełną funkcjonalność relacyjnej bazy danych, wsparcie dla złożonych zapytań, transakcje ACID i wysoką wydajność. Jest standardem w aplikacjach produkcyjnych, gdzie potrzebujesz niezawodności, skalowalności i zaawansowanych funkcji. Do pracy z PostgreSQL w Pythonie używa się biblioteki psycopg2, która oferuje zarówno niskopoziomowy interfejs, jak i wygodne abstrakcje. Poniżej znajdziesz przykład właściwego zarządzania połączeniami:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
import psycopg2
from psycopg2.extras import RealDictCursor
from contextlib import contextmanager

@contextmanager
def get_postgres_connection(host: str, database: str, user: str, password: str):
    """Context manager dla połączenia z PostgreSQL."""
    conn = psycopg2.connect(
        host=host,
        database=database,
        user=user,
        password=password
    )
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

# Przykład użycia PostgreSQL
with get_postgres_connection("localhost", "mydb", "user", "password") as conn:
    with conn.cursor(cursor_factory=RealDictCursor) as cursor:
        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        user = cursor.fetchone()
        print(dict(user))

Transakcje i bezpieczeństwo

Właściwe użycie transakcji:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def safe_database_operation(db_connection, operations: list):
    """Bezpieczne wykonywanie wielu operacji w transakcji."""
    try:
        cursor = db_connection.cursor()

        for operation in operations:
            cursor.execute(operation["sql"], operation.get("params", ()))

        db_connection.commit()
        return True
    except Exception as e:
        db_connection.rollback()
        print(f"Błąd transakcji: {e}")
        return False

# Przykład
operations = [
    {"sql": "INSERT INTO users (name, email) VALUES (?, ?)", "params": ("Jan", "jan@example.com")},
    {"sql": "UPDATE accounts SET balance = balance - 100 WHERE user_id = ?", "params": (1,)},
    {"sql": "INSERT INTO transactions (user_id, amount) VALUES (?, ?)", "params": (1, -100)}
]

with get_db_connection("bank.db") as conn:
    safe_database_operation(conn, operations)

Pipeline'y ETL

Ekstrakcja, transformacja, ładowanie:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import pandas as pd
from sqlalchemy import create_engine

def etl_pipeline(source_file: str, target_db: str, table_name: str):
    """Prosty pipeline ETL: Extract, Transform, Load."""
    # Extract: wczytanie danych
    df = pd.read_csv(source_file)
    print(f"Wczytano {len(df)} wierszy z {source_file}")

    # Transform: czyszczenie i transformacja
    df = df.dropna()
    df.columns = df.columns.str.lower().str.replace(' ', '_')

    # Load: zapis do bazy
    engine = create_engine(f"sqlite:///{target_db}")
    df.to_sql(table_name, engine, if_exists='replace', index=False)
    print(f"Zapisano do tabeli {table_name} w bazie {target_db}")

etl_pipeline("dane_surowe.csv", "baza.db", "dane_przetworzone")

Migracje bazy danych

Zarządzanie schematem bazy:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
from pathlib import Path

class DatabaseMigrator:
    def __init__(self, db_path: str):
        self.db_path = db_path
        self.migrations_dir = Path("migrations")
        self.migrations_dir.mkdir(exist_ok=True)

    def create_migration(self, name: str, sql: str):
        """Tworzy nową migrację."""
        migration_file = self.migrations_dir / f"{name}.sql"
        with open(migration_file, 'w') as f:
            f.write(sql)
        print(f"Utworzono migrację: {migration_file}")

    def run_migrations(self):
        """Uruchamia wszystkie migracje."""
        with get_db_connection(self.db_path) as conn:
            # Tabela śledząca migracje
            conn.execute("""
                CREATE TABLE IF NOT EXISTS migrations (
                    id INTEGER PRIMARY KEY,
                    name TEXT UNIQUE,
                    applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            """)

            # Uruchamianie migracji
            applied = {row[0] for row in conn.execute("SELECT name FROM migrations")}

            for migration_file in sorted(self.migrations_dir.glob("*.sql")):
                if migration_file.stem not in applied:
                    with open(migration_file) as f:
                        sql = f.read()
                    conn.executescript(sql)
                    conn.execute(
                        "INSERT INTO migrations (name) VALUES (?)",
                        (migration_file.stem,)
                    )
                    print(f"Zastosowano migrację: {migration_file.stem}")

migrator = DatabaseMigrator("baza.db")
migrator.run_migrations()

Podsumowanie

Praca z bazami danych w automatyzacji obejmuje:

• SQLite dla lokalnych zastosowań
• PostgreSQL dla produkcyjnych systemów
• Transakcje dla bezpieczeństwa danych
• Pipeline'y ETL do przetwarzania danych
• Migracje do zarządzania schematem

Te techniki są podstawą automatyzacji bazodanowej w Pythonie.

---

➡️ Następny artykuł

Po opanowaniu baz danych, naucz się automatyzować integracje z zewnętrznymi systemami:

Automatyzacja API i integracje w Pythonie — praktyczne wzorce pracy z REST API, obsługa błędów, retry, rate limiting i bezpieczne zarządzanie kluczami API.