Psycopg: connections, transactions, repository і tests

Connection — ресурс із явним lifecycle

Application отримує DSN із environment або secret store, а не з repository. Connection і cursor закривають через context managers або try/finally. Pool може бути доречним для server workload, але не потрібен автоматично для короткого CLI; спершу опишіть concurrency і lifetime.

import os
import psycopg

dsn = os.environ["APP_DATABASE_DSN"]
with psycopg.connect(dsn) as connection:
    with connection.cursor() as cursor:
        cursor.execute("SELECT id, name FROM python_sql_lab.projects ORDER BY id")
        rows = cursor.fetchall()

DSN не логують, не додають у ZIP і не вставляють у traceback report.

Transaction — business boundary

У Psycopg звичайна операція може почати transaction автоматично. Commit робить changes durable, rollback скасовує їх у поточній transaction. Після SQL error connection лишається в failed transaction state до rollback; продовжувати queries без очищення не можна.

Не тримайте transaction відкритою під час зовнішнього HTTP call або очікування user input: це збільшує contention і ризик stale state.

Autocommit — спеціальний режим

Autocommit не означає «автоматично безпечно». Він доречний для statements, які мають виконуватися поза transaction block, або окремих read/admin scenarios. Для multi-step business change потрібна explicit atomic transaction. Isolation level і read/write mode обирають за concurrency contract, а retry роблять лише для визначених transient failures та idempotent operation.

Repository відділяє SQL від use case

Repository приймає connection/cursor dependency, виконує parameterized SQL і повертає domain-friendly value. Він не читає глобальні environment variables, не робить commit потай і не перетворює кожну DB exception на None. Transaction owner залишається service/use case, який бачить весь operation.

def close_ticket(cursor, ticket_id: int) -> bool:
    cursor.execute(
        "UPDATE python_sql_lab.tickets SET status = %s "
        "WHERE id = %s AND status = %s",
        ("closed", ticket_id, "open"),
    )
    return cursor.rowcount == 1

rowcount дає доказ transition: zero може означати missing або вже closed, і цей domain choice треба зафіксувати.

Testing pyramid для database boundary

Pure mapping і validation тестуються без БД. Query contract-тести перевіряють placeholders, identifiers і transaction ownership. Інтеграційні tests мають запускатися на disposable PostgreSQL instance з окремою schema/database, migrations і deterministic seed; вони не повинні торкатися production.

Практичний pack цього модуля проходить deterministic contract tests локально; реальний PostgreSQL integration run учень виконує у власному disposable environment за інструкцією.

Методичні джерела

• Psycopg: Basic usage
• Psycopg: Transactions
• Python DB-API 2.0
• PostgreSQL: SET TRANSACTION

Урок, сценарії, пояснення й вправи створені SEOWORK. Посилання ведуть лише на офіційну документацію та Python Wiki.