Базы данных

Алгоритм работы с СУБД

Python
  │
  ▼
psycopg2.connect()
  │
  ▼
Соединение (connection)
  │
  ▼
Курсор (cursor)
  │
  ▼
SQL-запрос (execute)
  │
  ▼
Получение результата (fetch)
  │
  ▼
Закрытие курсора
  │
  ▼
Закрытие соединения

connection - соединение Python с PostgreSQL

cursor - выполнение SQL запросов и получение результатов

Аналогия:

connection — ты подключился по SSH к серверу.
cursor — открыл терминал, через который выполняешь команды.
execute() — отправил команду.
fetchall() — получил результат.
close() — закрыл соответствующий ресурс.

Подключение к СУБД

import psycopg2

conn = psycopg2.connect(
    host="192.168.1.100",
    port=5432,
    dbname="inventory",
    user="postgres",
    password="password"
)

cursor = conn.cursor()

cursor.execute("SELECT * FROM servers")

rows = cursor.fetchall()

for row in rows:
    print(row)

cursor.close()
conn.close()

Правильный способ подключения к базе

import psycopg2
from contextlib import closing

with closing(
    psycopg2.connect(
        host="192.168.1.100",
        port=5432,
        dbname="inventory",
        user="postgres",
        password="password"
    )
) as conn:

    with conn:
        with conn.cursor() as cursor:

            cursor.execute("SELECT * FROM servers")

            rows = cursor.fetchall()

            for row in rows:
                print(row)

Получение названий столбцов результата SQL-запроса

import psycopg2

conn = psycopg2.connect(
    host="localhost",
    database="mydb",
    user="postgres",
    password="password"
)

cur = conn.cursor()

cur.execute("SELECT * FROM servers")

for i, column in enumerate(cur.description, 1):
    print(f"{i}. {column.name}")

Как правильно выполнять SELECT

Например есть таблица:

servers

id | hostname | status
---+----------+--------
1  | server01 | active
2  | server02 | active
3  | server03 | failed

import psycopg2
from contextlib import closing

with closing(psycopg2.connect(
    host="localhost",
    dbname="inventory",
    user="postgres",
    password="password"
)) as conn:

    with conn:
        with conn.cursor() as cursor:

            cursor.execute(
                "SELECT id, hostname, status FROM servers"
            )

            for row in cursor.fetchall():
                print(row)

Результат:

(1, 'server01', 'active')
(2, 'server02', 'active')
(3, 'server03', 'failed')

Каждая строка представлена tuple (кортежем)

Обращение к значениям:

for row in cursor.fetchall():
    server_id = row[0]
    hostname = row[1]
    status = row[2]

    print(hostname, status)

Как получать строки в виде словарей

Вместо row[2] можно писать row["status"]

import psycopg2
from psycopg2.extras import RealDictCursor
from contextlib import closing

with closing(psycopg2.connect(
    host="localhost",
    dbname="inventory",
    user="postgres",
    password="password"
)) as conn:

    with conn:
        with conn.cursor(cursor_factory=RealDictCursor) as cursor:

            cursor.execute("SELECT * FROM servers")

            for row in cursor.fetchall():
                print(row["hostname"], row["status"])

INSERT, UPDATE и DELETE

with conn:
    with conn.cursor() as cursor:

        cursor.execute(
            """
            UPDATE servers
            SET status = %s
            WHERE hostname = %s
            """,
            ("failed", "server01")
        )

После успешного выхода из with conn: автоматически выполняется COMMIT.

Если SQL-запрос завершится ошибкой, произойдёт ROLLBACK.

Почему %s, а не f-строка?

Неправильно:

cursor.execute(

f"SELECT * FROM servers WHERE hostname = '{hostname}'"
)

Правильно:

cursor.execute(
"SELECT * FROM servers WHERE hostname = %s",
(hostname,)
)

Второй вариант обеспечивает безопасную передачу параметров и защищает от SQL-инъекций.

fetchone(), fetchmany(), fetchall()

Есть три основных способа получать данные.

Метод	Что делает	Когда использовать
`fetchone()`	Получает одну строку	Поиск одного сервера
`fetchmany(100)`	Получает до 100 строк	Пакетная обработка
`fetchall()`	Получает все оставшиеся строки	Небольшие выборки

Разница между close() и commit()

Команда	Назначение
`conn.commit()`	Подтверждает изменения транзакции
`conn.rollback()`	Откатывает изменения транзакции
`conn.close()`	Закрывает соединение
`cursor.close()`	Закрывает курсор
`with conn:`	Автоматически выполняет commit или rollback
`with conn.cursor():`	Автоматически закрывает курсор
`closing(conn)`	Закрывает соединение при выходе из блока

Предупреждение

Важно: conn.close() не заменяет conn.commit(). Незавершённые изменения при закрытии соединения не сохраняются.

Как обрабатывать ошибки

import psycopg2
from contextlib import closing

try:
    with closing(psycopg2.connect(
        host="localhost",
        dbname="inventory",
        user="postgres",
        password="password",
        connect_timeout=10
    )) as conn:

        with conn:
            with conn.cursor() as cursor:

                cursor.execute("SELECT * FROM servers")

                for row in cursor.fetchall():
                    print(row)

except psycopg2.OperationalError as error:
    print(f"Ошибка подключения: {error}")

except psycopg2.DatabaseError as error:
    print(f"Ошибка базы данных: {error}")

Здесь

OperationalError — например, PostgreSQL недоступен или соединение оборвалось.
DatabaseError — более общая категория ошибок работы с БД.
connect_timeout=10 — ограничение времени установления соединения десятью секундами.

Еще одна архитектура

import os
import psycopg2

from contextlib import closing
from psycopg2.extras import RealDictCursor


def get_connection():
    return psycopg2.connect(
        host="localhost",
        port=5432,
        dbname="inventory",
        user="postgres",
        password=os.environ["PGPASSWORD"],
        connect_timeout=10
    )


def get_servers(conn):
    with conn.cursor(cursor_factory=RealDictCursor) as cursor:

        cursor.execute("""
            SELECT id, hostname, status
            FROM servers
        """)

        return cursor.fetchall()


def main():
    with closing(get_connection()) as conn:
        with conn:
            servers = get_servers(conn)

        for server in servers:
            print(server["hostname"], server["status"])


if __name__ == "__main__":
    main()

Последняя проверка 11.10.2026