Обработка транзакций с помощью Psycopg2

Оригинал статьи: Transaction Handling with Psycopg2

Дата публикации: 6 декабря 2017

Автор: Benjamin Bengfort

Базы данных необходимы для большинства приложений, однако большая часть взаимодействия с базами данных часто игнорируется разработчиками Python, которые используют библиотеки более высокого уровня, такие как Django или SQLAlchemy. Мы используем и любим PostgreSQL с Psycopg2, но недавно я понял, что не очень хорошо понимаю, как именно psycopg2 реализует основные концепции баз данных: в частности, изоляция транзакций и безопасность потоков.

Вот что говорится в документации относительно транзакций:

Транзакции обрабатываются классом соединения connection. По умолчанию при первой отправке команды в базу данных (с использованием одного из курсоров cursor, созданных соединением) создается новая транзакция. Следующие команды базы данных будут выполнены в контексте той же транзакции — не только команды, выданные первым курсором, но и те, которые выданы всеми курсорами, созданными тем же соединением. Если какая-либо команда не будет выполнена, транзакция будет прервана, и никакая следующая команда не будет выполнена до вызова метода rollback().

Таким образом, транзакции зависят от соединения. Когда вы создаете соединение, вы можете создать несколько курсоров, транзакция начинается, когда первый курсор выдает команду execute – все команды, выполняемые всеми курсорами после этого, являются частью одной транзакции до фиксации commit или отката rollback. После вызова любого из этих методов следующая транзакция запускается при следующем вызове execute.

Это поднимает очень важный момент:

По умолчанию даже простой SELECT начнет транзакцию: в долго работающих программах, если не предпринимать никаких дальнейших действий, сеанс останется «простаивающим в транзакции» (idle in transaction, что нежелательно по нескольким причинам (сеанс удерживает блокировки, таблицы раздуваются…). Для долгоживущих скриптов либо убедитесь, что вы завершаете транзакцию как можно скорее, либо используйте соединение с автофиксацией autocommit.

Похоже, это указывает на то, что при работе напрямую с psycopg2 понимание транзакций необходимо для написания стабильных скриптов. Поэтому в этой статье подробно описаны мои заметки и методы для более эффективной работы с PostgreSQL из Python.

Предварительные сведения о базе данных

Чтобы продемонстрировать код в этой статье блога, нам нужна база данных. Классический пример базы данных, который преподают студентам, — это банковский счет, поэтому мы продолжим эту тему здесь! Извините, если эта часть утомительна, можете пропустить ее. В файле schema.sql я определил следующую схему как DDL (язык определения данных):

DROP TABLE IF EXISTS users CASCADE;
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    username VARCHAR(255) UNIQUE,
    pin SMALLINT NOT NULL
);


DROP TYPE IF EXISTS account_type CASCADE;
CREATE TYPE account_type AS ENUM ('checking', 'savings');

DROP TABLE IF EXISTS accounts CASCADE;
CREATE TABLE accounts (
    id SERIAL PRIMARY KEY,
    type account_type,
    owner_id INTEGER NOT NULL,
    balance NUMERIC DEFAULT 0.0,
    CONSTRAINT positive_balance CHECK (balance >= 0),
    FOREIGN KEY (owner_id) REFERENCES users (id)
);

DROP TYPE IF EXISTS ledger_type CASCADE;
CREATE TYPE ledger_type AS ENUM ('credit', 'debit');

DROP TABLE IF EXISTS ledger;
CREATE TABLE ledger (
    id SERIAL PRIMARY KEY,
    account_id INTEGER NOT NULL,
    date DATE NOT NULL DEFAULT CURRENT_DATE,
    type ledger_type NOT NULL,
    amount NUMERIC NOT NULL,
    FOREIGN KEY (account_id) REFERENCES accounts (id)
);

Это создает простую базу данных с двумя таблицами. Таблица владельцев содержит PIN-код для проверки. Владельцы могут иметь один или несколько счетов, а счета имеют ограничение, что баланс никогда не может быть ниже $0.00. Мы также можем заполнить базу данных некоторыми начальными данными:

INSERT INTO users (id, username, pin) VALUES
    (1, 'alice', 1234),
    (2, 'bob', 9999);

INSERT INTO accounts (type, owner_id, balance) VALUES
    ('checking', 1, 250.0),
    ('savings', 1, 5.00),
    ('checking', 2, 100.0),
    ('savings', 2, 2342.13);

Переходя к коду Python, мы можем добавить некоторый шаблонный код, который позволит нам подключиться к базе данных и выполнить SQL в нашем файле выше:

import os
import psycopg2 as pg


def connect(env="DATABASE_URL"):
    url = os.getenv(env)
    if not url:
        raise ValueError("no database url specified")
    return pg.connect(url)


def createdb(conn, schema="schema.sql"):
    with open(schema, 'r') as f:
        sql = f.read()

    try:
        with conn.cursor() as curs:
            curs.execute(sql)
            conn.commit()
    except Exception as e:
        conn.rollback()
        raise e

Функция connect ищет строку подключения к базе данных в переменной среды $DATABASE_URL. Поскольку код конфигурации базы данных может содержать пароли и сетевую информацию, всегда лучше хранить его в среде или в локальном, защищенном файле конфигурации, к которому может получить доступ только процесс и который не может быть зарегистрирован с помощью кода. Строка подключения должна выглядеть примерно так: postgresql://user@localhost:5432/dbname.

Функция createdb считывает SQL из файла schema.sql и выполняет его в базе данных. Обратите внимание, что именно поэтому у нас есть операторы DROP TABLE IF EXISTS, чтобы мы могли гарантировать, что всегда начинаем с новой базы данных при запуске этого скрипта. Эта функция также дает нам первый взгляд на транзакции и взаимодействие базы данных с Python.

Соответствуя PEP 249, мы создаем подключение к базе данных, затем создаем курсор из подключения. Курсоры управляют выполнением SQL в отношении базы данных, а также извлечением данных. Мы выполняем SQL в нашем файле схемы, фиксируя транзакцию, если не возникает никаких исключений, и откатывая ее, если она не выполняется. Мы рассмотрим это подробнее в следующем разделе.

Управление транзакциями

Транзакция состоит из одной или нескольких связанных операций, которые представляют собой одну единицу работы. Например, в примере с банковским счетом у вас может быть транзакция депозита, которая выполняет запросы для поиска счета и проверки пользователя, добавления записи в список ежедневных депозитов, проверки достижения ежедневного лимита депозита, а затем изменения баланса счета. Все эти операции представляют собой все шаги, необходимые для выполнения депозита.

Цель транзакции заключается в том, чтобы после завершения транзакции база данных оставалась в едином согласованном состоянии. Согласованность часто определяется инвариантами или ограничениями, которые на более высоком уровне описывают, как база данных должна хранить информацию. С точки зрения программирования, если эти ограничения нарушаются, возникает исключение. Например, база данных имеет ограничение positive_balance, если баланс счета становится ниже нуля, возникает исключение. При нарушении этого ограничения база данных должна оставаться неизменной, а все операции, выполненные транзакцией, должны быть отменены (rollback). Если транзакция прошла успешно, мы можем зафиксировать изменения (commit), что гарантирует, что база данных успешно применила нашу операцию.

Так зачем же нам нужно управлять транзакциями? Рассмотрим следующий код:

conn = connect()
curs = conn.cursor()

try:
    # Выполнить команду, которая вызовет ограничение
    curs.execute("UPDATE accounts SET balance=%s", (-130.935,))
except Exception as e:
    print(e) # Исключение ограничения

# Выполнить еще одну команду, но из-за предыдущего исключения:
curs = conn.cursor()
try:
    curs.execute("SELECT id, type FROM accounts WHERE owner_id=%s", (1,))
except pg.InternalError as e:
    print(e)

Первый curs.execute запускает исключение ограничения, которое перехватывается и выводится. Однако база данных теперь находится в несогласованном состоянии. При попытке выполнить второй запрос возникает psycopg2.InternalError: "current transaction is aborted, commands ignored until end of transaction block". Чтобы продолжить работу с приложением, необходимо вызвать conn.rollback(), чтобы завершить транзакцию и начать новую.

Использование conn.cursor() в качестве curs: приводит к тому же поведению, менеджер контекста не очищает состояние транзакции автоматически.

По сути, это означает, что все транзакции можно обернуть в блок try, если они завершаются успешно, их можно зафиксировать, однако если они вызывают исключение, их необходимо откатить. Базовый декоратор, который делает это, выглядит следующим образом:

from functools import wraps

def transaction(func):
    @wraps(func)
    def inner(*args, **kwargs):
        conn = connect()
        try:
            func(conn, *args, **kwargs)
            conn.commit()
        except Exception as e:
            conn.rollback()
            log.error("{} error: {}".format(func.__name__, e))
        finally:
            conn.close()
    return inner

Этот декоратор оборачивает указанную функцию, возвращая внутреннюю функцию, которая внедряет новое соединение в качестве первого аргумента в декорированную функцию. Если декорированная функция вызывает исключение, транзакция откатывается, а ошибка регистрируется.

Метод декоратора хорош, но внедрение соединения может быть немного странным. Альтернативой является менеджер контекста, который гарантирует, что соединение зафиксировано или откатится аналогичным образом:

from contextlib import contextmanager

@contextmanager
def transaction():
    try:
        conn = connect()
        yield conn
        conn.commit()
    except Exception as e:
        conn.rollback()
        log.error("db error: {}".format(e))
    finally:
        conn.close()

Это позволяет вам писать код с использованием with следующим образом:

with transaction() as conn:
    # транзакция

Контекстный менеджер позволяет вам легко объединить две транзакции внутри одной функции — конечно, это может быть не по теме. Однако не составит труда объединить методы декоратора и контекстного менеджера в два шага (подробнее об этом в уровнях изоляции).

Приложение ATM

Итак, давайте поговорим о двух конкретных транзакциях для воображаемого приложения базы данных: депозит и снятие. Каждая из этих операций состоит из нескольких шагов:

1. Подтвердите пользователя с помощью связанного PIN-кода

2. Убедитесь, что пользователь владеет изменяемым счетом

3. Запишите запись в бухгалтерскую книгу с применением кредита или дебета

4. В кредите убедитесь, что ежедневный лимит депозита не достигнут

5. Измените баланс счета

6. Извлеките текущий баланс для отображения пользователю

Каждая транзакция будет выполнять 6-7 различных SQL-запросов: SELECT, INSERT и UPDATE. Если какой-либо из них не сработает, то база данных должна остаться полностью неизменной. В этом случае сбой заключается в том, что возникает исключение, что потенциально является самым простым способом, когда у вас есть стек функций, вызывающих другие функции. Давайте сначала рассмотрим deposit:

@transaction
def deposit(conn, user, pin, account, amount):
    # Шаг 1: аутентифицируйте пользователя с помощью PIN-кода
    # и подтвердите право собственности на аккаунт
    authenticate(conn, user, pin, account)

    # Шаг 2: добавьте запись в бухгалтерскую книгу с кредитом
    ledger(conn, account, "credit", amount)

    # Шаг 3: обновите значение счета, добавив сумму
    update_balance(conn, account, amount)

    # Получите текущий баланс на счете и запишите его.
    record = "withdraw ${:0.2f} from account {} | current balance: ${:0.2f}"
    log.info(record.format(amount, account, balance(conn, account)))

Эта функция просто вызывает другие функции, передавая контекст транзакции (в данном случае соединение, а также входные данные) другим функциям, которые могут или не могут вызывать исключения. Вот два метода аутентификации:

def authenticate(conn, user, pin, account=None):
    """
    Возвращает идентификатор учетной записи, если имя найдено и пин-код совпадает.
    """
    with conn.cursor() as curs:
        sql = "SELECT 1 AS authd FROM users WHERE username=%s AND pin=%s"
        curs.execute(sql, (user, pin))
        if curs.fetchone() is None:
            raise ValueError("could not validate user via PIN")
        return True

    if account:
        # Подтвердите право собственности на учетную запись, если она предоставлена
        verify_account(conn, user, account)


def verify_account(conn, user, account):
    """
    Убедитесь, что учетная запись принадлежит пользователю.
    """
    with conn.cursor() as curs:
        sql = (
            "SELECT 1 AS verified FROM accounts a "
            "JOIN users u on u.id = a.owner_id "
            "WHERE u.username=%s AND a.id=%s"
        )
        curs.execute(sql, (user, account))

        if curs.fetchone() is None:
            raise ValueError("account belonging to user not found")
        return True

Функции authenticate и verify_account в основном ищут в базе данных запись, которая соответствует условиям — пользователь с соответствующим PIN-кодом в authenticate и пара (user, account_id) в verify_account. Обе эти функции полагаются на ограничение UNIQUE в базе данных для имен пользователей и идентификаторов учетных записей. Этот пример показывает, как стек вызовов функций может стать произвольно глубоким; verify_account вызывается authenticate, который вызывается deposit. При возникновении исключения в любой точке стека транзакция не будет продолжена, что защищает нас от вреда в дальнейшей транзакции.

Обратите внимание также, что ни одна из этих функций не имеет декоратора @transaction, это связано с тем, что ожидается, что они будут вызваны из другой транзакции. Это независимые операции, но их можно вызывать независимо в транзакции с помощью менеджера контекста.

Далее мы вставляем запись в реестр:

MAX_DEPOSIT_LIMIT = 1000.00

def ledger(conn, account, record, amount):
    """
    Добавьте запись в бухгалтерскую книгу с указанием суммы,
    которая кредитуется или дебетуется.
    """
    # Выполнить вставку
    with conn.cursor() as curs:
        sql = "INSERT INTO ledger (account_id, type, amount) VALUES (%s, %s, %s)"
        curs.execute(sql, (account, record, amount))

    # Если мы зачисляем средства на счет, выполняйте ежедневную проверку депозита.
    if record == "credit":
        check_daily_deposit(conn, account)

def check_daily_deposit(conn, account):
    """
    Создайте исключение, если лимит депозита был превышен.
    """
    with conn.cursor() as curs:
        sql = (
            "SELECT amount FROM ledger "
            "WHERE date=now()::date AND type='credit' AND account_id=%s"
        )
        curs.execute(sql, (account,))
        total = sum(row[0] for row in curs.fetchall())
        if total > MAX_DEPOSIT_LIMIT:
            raise Exception("daily deposit limit has been exceeded!")

Это первое место, где мы изменяем состояние базы данных, вставляя запись в реестр. Если при проверке check_daily_deposit мы обнаруживаем, что наш лимит депозита был превышен за день, возникает исключение, которое откатывает транзакцию. Это гарантирует, что запись в реестре не будет случайно сохранена на диске. Наконец, мы обновляем баланс счета:

def update_balance(conn, account, amount):
    """
    Добавьте сумму (или вычтите, если она отрицательная) к остатку на счете.
    """
    amount = Decimal(amount)
    with conn.cursor() as curs:
        current = balance(conn, account)
        sql = "UPDATE accounts SET balance=%s WHERE id=%s"
        curs.execute(sql, (current+amount, account))


def balance(conn, account):
    with conn.cursor() as curs:
        curs.execute("SELECT balance FROM accounts WHERE id=%s", (account,))
        return curs.fetchone()[0]

Я еще расскажу об update_balance, когда мы обсудим уровни изоляции, но достаточно сказать, что это еще одно место, где в случае сбоя транзакции мы хотим убедиться, что наш счет не будет изменен! Чтобы завершить пример, вот транзакция вывода withdraw:

@transaction
def withdraw(conn, user, pin, account, amount):
    # Шаг 1: аутентифицируйте пользователя с помощью PIN-кода
    # и подтвердите право собственности на аккаунт
    authenticate(conn, user, pin, account)

    # Шаг 2: добавьте запись в бухгалтерскую книгу с дебетом
    ledger(conn, account, "debit", amount)

    # Шаг 3: обновите значение счета, вычтя сумму
    update_balance(conn, account, amount * -1)

    # Получите текущий баланс на счете и запишите его.
    record = "withdraw ${:0.2f} from account {} | current balance: ${:0.2f}"
    log.info(record.format(amount, account, balance(conn, account)))

Это похоже, но изменяет входные данные для различных операций, чтобы уменьшить сумму счета на запись в дебетовой книге. Мы можем запустить:

if __name__ == '__main__':
    conn = connect()
    createdb(conn)

    # Успешный депозит
    deposit('alice', 1234, 1, 785.0)

    # Успешный вывод
    withdraw('alice', 1234, 1, 230.0)

    # Ошибочный депозит
    deposit('alice', 1234, 1, 489.0)

    # Успешный депозит
    deposit('bob', 9999, 2, 220.23)

И мы должны увидеть следующие записи журнала:

2017-12-06 20:01:00,086 withdraw $785.00 from account 1 | current balance: $1035.00
2017-12-06 20:01:00,094 withdraw error: could not validate user via PIN
2017-12-06 20:01:00,103 withdraw $230.00 from account 1 | current balance: $805.00
2017-12-06 20:01:00,118 deposit error: daily deposit limit has been exceeded!
2017-12-06 20:01:00,130 withdraw $220.23 from account 2 | current balance: $225.23

Это должно задать базовый уровень для создания простых и удобных в использовании транзакций в Python. Однако, если вы помните свой курс по базам данных в бакалавриате, все становится интереснее, когда две транзакции происходят одновременно. Мы рассмотрим это на примере одного процесса, рассмотрев многопоточные соединения с базой данных.

Потоки

Давайте рассмотрим, как запустить две транзакции одновременно из одного приложения. Самый простой способ сделать это — использовать библиотеку потоков для одновременного выполнения транзакций. Как добиться безопасности потоков при доступе к базе данных? Возвращаясь к документации:

Объекты соединения потокобезопасны: множество потоков могут обращаться к одной и той же базе данных, используя отдельные сеансы и создавая соединение на поток, или используя одно и то же соединение и создавая отдельные курсоры. На языке DB API 2.0 Psycopg является потокобезопасным на уровне 2.

Это означает, что каждый поток должен иметь свой собственный объект conn (который исследуется в разделе пула подключений). Любой курсор, созданный из одного и того же объекта подключения, будет находиться в одной транзакции независимо от потока. Мы также хотим рассмотреть, как каждая транзакция влияет друг на друга, и мы сначала рассмотрим это, исследуя уровни изоляции и состояние сеанса.

Состояние сессии

Допустим, у Алисы и Чарли есть общий счет на имя Алисы. Они оба появляются у банкоматов в одно и то же время, Алиса пытается положить 75 долларов, а затем снять 25 долларов, а Чарли пытается снять 300 долларов. Мы можем смоделировать это с помощью потоков следующим образом:

import time
import random
import threading

def op1():
    time.sleep(random.random())
    withdraw('alice', 1234, 1, 300.0)

def op2():
    time.sleep(random.random())
    deposit('alice', 1234, 1, 75.0)
    withdraw('alice', 1234, 1, 25.0)


threads = [
    threading.Thread(target=op1),
    threading.Thread(target=op2),
]

for t in threads:
    t.start()

for t in threads:
    t.join()

В зависимости от времени может произойти одно из двух. Чарли может быть отклонен из-за отсутствия достаточного количества денег на счете, и конечное состояние базы данных может быть 300 долларов, или все транзакции могут быть успешными с конечным состоянием базы данных, установленным на 0 долларов. Происходит три транзакции, две транзакции снятия withdraw и депозит deposit. Каждая из этих транзакций выполняется изолированно, то есть они видят базу данных, как они начинались, и любые изменения, которые они вносят; поэтому, если снятие withdraw Чарли и депозит deposit Алисы происходят одновременно, Чарли будет отклонен, так как он не знает о депозите, пока он не будет завершен. Независимо от того, что происходит, база данных останется в том же состоянии.

Однако из соображений производительности вы можете изменить уровень изоляции для конкретной транзакции. Возможны следующие уровни:

1. READ UNCOMMITTED: самый низкий уровень изоляции, транзакция может считывать значения, которые еще не зафиксированы (и могут никогда не быть зафиксированы).

2. READ COMMITTED: блокировки записи сохраняются, но блокировки чтения снимаются после выбора, что означает, что два разных значения могут быть прочитаны в разных частях транзакции.

3. REPEATABLE READ: сохранять блокировки чтения и записи, чтобы несколько чтений возвращали одинаковые значения, но могут происходить фантомные чтения.

4. SERIALIZABLE: самый высокий уровень изоляции: блокировки чтения, записи и диапазона сохраняются до конца транзакции.

5. DEFAULT: устанавливается конфигурацией сервера, а не Python, обычно READ COMMITTED.

Обратите внимание, что по мере увеличения уровня изоляции увеличивается и количество поддерживаемых блокировок, что серьезно влияет на производительность в случае возникновения конфликтов блокировок или взаимоблокировок. Можно установить уровень изоляции для каждой транзакции, чтобы повысить производительность всех транзакций, происходящих одновременно. Для этого мы должны изменить параметры сеанса в соединении, которые изменяют поведение транзакции или операторов, следующих в этом конкретном сеансе. Кроме того, мы можем установить сеанс только для чтения readonly, что не позволяет выполнять запись во временные таблицы (для производительности и безопасности) или отложить deferrable.

Отсрочка очень интересна в транзакции, поскольку она изменяет способ проверки ограничений базы данных. Неотсроченные транзакции немедленно проверяют ограничение после выполнения оператора. Это означает, что UPDATE accounts SET balance=-5.45 немедленно вызовет исключение. Однако отсроченные транзакции ждут завершения транзакции перед проверкой ограничений. Это позволяет вам писать несколько перекрывающихся операций, которые могут привести базу данных в правильное состояние к концу транзакции, но потенциально не во время транзакции (это также перекрывается с производительностью различных уровней изоляции).

Чтобы изменить сеанс, мы будем использовать менеджер контекста, как мы делали раньше, чтобы изменить сеанс для транзакции, а затем сбросим сеанс обратно к значениям по умолчанию:

@contextmanager
def session(conn, isolation_level=None, readonly=None, deferrable=None):
    try:
        conn.set_session(
            isolation_level=isolation_level,
            readonly=readonly,
            deferrable=deferrable
        )
        yield conn
    finally:
        # Сбросить сеанс к настройкам по умолчанию
        conn.set_session(None, None, None, None)

Затем мы можем использовать with для проведения транзакций с различными уровнями изоляции:

with transaction() as conn:
    with session(conn, isolation_level="READ COMMITTED") as conn:
        # транзакция

Не может быть текущей транзакции, когда сеанс установлен, поэтому я чаще устанавливаю уровень изоляции, только для чтения и отложенный внутри декоратора транзакции, а не использую два отдельных менеджера контекста, как показано выше. Честно говоря, также часто эти свойства устанавливаются на основе процесса, а не на основе транзакции, поэтому сеанс устанавливается в connect.

Пулы подключений

Подключения не могут быть общими для потоков. В примере потоков выше, если мы удалим декоратор @transaction и передадим одно и то же подключение в обе операции следующим образом:

conn = connect()

def op1():
    time.sleep(random.random())
    withdraw(conn, 'alice', 1234, 1, 300.0)

def op2():
    time.sleep(random.random())
    deposit(conn, 'alice', 1234, 1, 75.0)
    withdraw(conn, 'alice', 1234, 1, 25.0)

Если op1 withdraw срабатывает первым, исключение приведет к тому, что все операторы op2 также не сработают, поскольку они находятся в одной транзакции. По сути, это означает, что и op1, и op2 находятся в одной транзакции, хотя они находятся в разных потоках!

До сих пор мы избегали этого, создавая новое соединение каждый раз, когда выполняется транзакция. Однако подключение к базе данных может быть дорогим, и при высоких транзакционных нагрузках мы можем просто оставить соединение открытым, но гарантировать, что они используются только одной транзакцией за раз. Решение заключается в использовании пулов соединений. Мы можем изменить нашу функцию соединения следующим образом:

from psycopg2.pool import ThreadedConnectionPool

def connect(env="DATABASE_URL", connections=2):
    """
    Подключитесь к базе данных, используя переменную среды.
    """
    url = os.getenv(env)
    if not url:
        raise ValueError("no database url specified")

    minconns = connections
    maxconns = connections * 2
    return ThreadedConnectionPool(minconns, maxconns, url)

Это создает потокобезопасный пул соединений, который устанавливает не менее 2 соединений и будет увеличиваться до максимум 4 соединений по требованию. Чтобы использовать объект пула в нашем декораторе транзакций, нам придется подключиться, когда декоратор импортируется, создавая глобальный объект пула:

pool = connect()

@contextmanager
def transaction(name="transaction", **kwargs):
    # Получить параметры сеанса из kwargs
    options = {
        "isolation_level": kwargs.get("isolation_level", None),
        "readonly": kwargs.get("readonly", None),
        "deferrable": kwargs.get("deferrable", None),
    }

    try:
        conn = pool.getconn()
        conn.set_session(**options)
        yield conn
        conn.commit()
    except Exception as e:
        conn.rollback()
        log.error("{} error: {}".format(name, e))
    finally:
        conn.reset()
        pool.putconn(conn)

Использование pool.getconn извлекает соединение из пула (если оно доступно, блокируя до тех пор, пока оно не будет готово), затем, когда мы закончим, мы можем pool.putconn освободить объект соединения.

Заключение

Это была куча заметок о более прямом использовании psycopg2. Извините, я не смог написать более убедительное заключение, но уже поздно, и этот пост теперь почти в 4 тыс. слов. Пора идти ужинать!

Заметки

Я использовал журналирование в качестве основного вывода для этого приложения. Журналирование было настроено следующим образом:

import logging

LOG_FORMAT = "%(asctime)s %(message)s"

logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
log = logging.getLogger('balance')

Полный код смотрите здесь.