PostgreSQL Python

Введение

В этой статье вы можете изучить примеры работы с PostgreSQL на языке программирования Python.

Прежде чем пытаться подключиться к Postgres убедитесь, что он установлен и запущен.

Пример установки Postgres 17 в Windows 11 вы можете изучить здесь

Простое подключение с psycopg2

Рассмотрим простой пример подключения к базе данных с помощью psycopg2

Учётные данные будем хранить в файле database.ini читать из него будем скриптом config.py на основе configparser а само подлючение реализуем в pg_demo.py

pg_demo/ |-- .gitignore |-- config.py |-- database.ini `-- pg_demo.py

# .gitignore database.ini __pycache__

В файле database.ini самым важным является поле password. Нужно указать тот пароль, который был использован при установке Postgres.

Порт по умолчанию 5432 можно не указывать, но если серверов несколько они будут на разных портах.

[postgresql] host=localhost port=5432 database=postgres user=postgres password=secret

#!/usr/bin/python # config.py import os from configparser import ConfigParser def config(filename="database.ini", section="postgresql"): # create a parser parser = ConfigParser() file_path = os.path.join(os.path.dirname(__file__), filename) # read config file parser.read(file_path) # get section, default to postgresql db = {} if parser.has_section(section): params = parser.items(section) for param in params: db[param[0]] = param[1] else: raise Exception(f"Section {section} not found in the {filename} file") return db

#!/usr/bin/python import psycopg2 from config import config def connect(): """ Connect to the PostgreSQL database server """ conn = None try: # read connection parameters params = config() # connect to the PostgreSQL server print('Connecting to the PostgreSQL database...') conn = psycopg2.connect(**params) # create a cursor cur = conn.cursor() # execute a statement print('PostgreSQL database version:') cur.execute('SELECT version()') # display the PostgreSQL database server version db_version = cur.fetchone() print(db_version) # close the communication with the PostgreSQL cur.close() except (Exception, psycopg2.DatabaseError) as error: print(error) finally: if conn is not None: conn.close() print('Database connection closed.') if __name__ == '__main__': connect()

python pg_demo.py

Connecting to the PostgreSQL database... PostgreSQL database version: ('PostgreSQL 12.8, compiled by Visual C++ build 1914, 64-bit',) Database connection closed.

Создать БД

Для создания базы данных используем сетап из предыдущего примера. Добавим в него файл create_db.py

Подключаться будем с помощью config.py

Обратим внимание, на то, что для создания новой базы данных мы подлючаемся к дефолтной БД postgres. После создания новой базы для работы с ней нужно будет подключаться уже непосредственно к ней.

Если вы используете только что установленный Postgres, в pgAdmin он будет выглядеть так

Создадим базу my_db

# ctreate_db.py import psycopg2 from psycopg2 import sql from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT from config import config params = config() con = psycopg2.connect(**params) con.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT) cur = con.cursor() cur.execute(sql.SQL("CREATE DATABASE {}").format(sql.Identifier("my_db"))) cur.close() con.close() print("Database created successfully.")

python create_db.py

Database created successfully

Создать таблицу

Чтобы создать таблицу в нашей новой базе данных my_db сначала создадим для неё секцию с настройками в database.ini

[my_db] host=localhost port=5432 database=my_db user=postgres password=secret

Создадим новый скрипт create_table.py

Подключаться будем по-прежнему с помощью config.py , но теперь явно укажем секцию.

В pgAdmin мы видим, что база данных пока пустая

# create_table.py import psycopg2 from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT from config import config params = config(filename="database.ini", section="my_db") con = psycopg2.connect(**params) con.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT) cur = con.cursor() create_table_query = ''' CREATE TABLE cities ( id SERIAL PRIMARY KEY, name TEXT, country TEXT, population INT, founded DATE, capital BOOL, visited BOOL, upd_timestamp BIGINT ); ''' cur.execute(create_table_query) cur.close() con.close() print("Table created successfully.")

python create_table.py

Table created successfully

В pgAdmin мы видим, что в базе данных my_db появилась пустая таблица cities

Добавить данные в таблицу

Создадим скрипт insert_to_table.py , который будет добавлять новые строки в таблицу cities базы данных my_db

Сперва рассмотрим классический способ добавления данных

# insert_to_table_classic.py import time import psycopg2 from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT from config import config from datetime import date def insert_to_table(table_name: str, data: tuple) -> None: params = config(filename="database.ini", section="my_db") con = psycopg2.connect(**params) con.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT) cur = con.cursor() cur.execute(f"INSERT INTO {table_name} " f"(name, " f"country, " f"population, " f"founded, " f"capital, " f"visited, " f"upd_timestamp) " f"VALUES (%s, %s, %s, %s, %s, %s, %s)", data) cur.close() con.close() print("Data inserted successfully.") data = ("Narva", "Estonia", 40000, date(1172, 1, 1), False, True, int(time.time())) if __name__ == "__main__": insert_to_table("cities", data)

python insert_to_table_classic.py

Data inserted successfully.

В pgAdmin мы видим, что в таблице cities появилась запись о Нарве

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

Добавить данные с помощью словаря

Существует способ добавлять данные в таблицу не указывая явно названия столбцов. Все нужные значения берутся из словаря . При использовании INSERT преимущество такого способа спорно. Гораздо лучше он раскрывается при обновлении таблицы

# insert_to_table.py import time import psycopg2 from psycopg2.extensions import AsIs from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT from config import config from datetime import date def insert_to_table(table_name: str, data: dict) -> None: params = config(filename="database.ini", section="my_db") con = psycopg2.connect(**params) con.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT) cur = con.cursor() data.update({"upd_timestamp": int(time.time())}) columns = data.keys() values = [data[column] for column in columns] insert_statement = f"INSERT INTO {table_name} (%s) VALUES %s" cur.execute(insert_statement, (AsIs(','.join(columns)), tuple(values))) cur.close() con.close() print("Data inserted successfully.") data = { "name": "Riga", "country": "Latvia", "population": 200000, "founded": date(1201, 1, 1 ), "capital": True, "visited": True } if __name__ == "__main__": insert_to_table("cities", data)

python insert_to_table.py

Data inserted successfully.

В pgAdmin мы видим, что в таблице cities появилась запись о Риге

Обновление таблицы

Обновим данные о Нарве классическим способом.

# update_table_classic.py import time import psycopg2 from datetime import date from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT from config import config def update_table(name: str, table_name: str, values: tuple) -> None: params = config(filename="database.ini", section="my_db") con = psycopg2.connect(**params) con.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT) cur = con.cursor() update_query = (f"UPDATE {table_name} SET " f"name = %s, " f"country = %s, " f"population = %s, " f"founded = %s, " f"capital = %s, " f"visited = %s, " f"upd_timestamp = %s " f"WHERE name = '{name}'") cur.execute(update_query, values) cur.close() con.close() print("Data updated successfully.") values = ("Narva", "Estonia", 53626, date(1172, 1, 1), False, True, int(time.time())) if __name__ == "__main__": update_table("Narva", "cities", values)

python update_table_classic.py

Data updated successfully

В pgAdmin мы видим, что обновились данные о численности населения и поле upd_timestamp

Обновление отдельных полей

В предыдущем примере были обновлены все поля. Некоторые были обновлены теми же самыми значениями, другие - новыми. Если нужно обновить только определённые поля нужно исключить лишние из запроса.

Рассмотрим пример обновления полей population и upd_timestamp классическим способом.

# update_table_classic.py import time import psycopg2 from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT from config import config def update_table(name: str, table_name: str, values: tuple) -> None: params = config(filename="database.ini", section="my_db") con = psycopg2.connect(**params) con.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT) cur = con.cursor() update_query = (f"UPDATE {table_name} SET " f"population = %s, " f"upd_timestamp = %s " f"WHERE name = '{name}'") cur.execute(update_query, values) cur.close() con.close() print("Data updated successfully.") values = (53627, int(time.time())) if __name__ == "__main__": update_table("Narva", "cities", values)

python update_table.py

Data updated successfully

В pgAdmin мы видим, что в таблице cities изменились численность населения Нарвы и время обновления

Если теперь нужно обновить другое поле, потребуется отдельная функция. Для каждой комбинации полей нужна отдельная функция для обновления. Это не очень удобно, поэтому можно использовать обновление из словаря.

Обновление из словаря

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

Создадим скрипт update_table.py , который будет обновлять только те значения, которые содержатся в словаре.

# update_table.py import time import psycopg2 from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT from config import config def update_table(name: str, table_name: str, data: dict) -> None: params = config(filename="database.ini", section="my_db") con = psycopg2.connect(**params) con.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT) cur = con.cursor() data.update({"upd_timestamp": int(time.time())}) set_clause = ', '.join([f"{key} = %({key})s" for key in data if key != 'id']) update_statement = f"UPDATE {table_name} SET {set_clause} WHERE name = '{name}'" cur.execute(update_statement, data) cur.close() con.close() print("Data updated successfully.") data = { "population": 619000, "visited": False } if __name__ == "__main__": update_table("Riga", "cities", data)

python update_table.py

Data updated successfully

В pgAdmin мы видим, что в таблице cities изменились численность населения Риги , статус посещения и временная отметка.

Обновление из словаря с помощью конкатенации

Аналогичного результата можно добиться с помощью манипуляций над строкой SQL запроса

# update_table.py import time import psycopg2 from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT from config import config def update_table(name: str, table_name: str, data: dict) -> None: params = config(filename="database.ini", section="my_db") con = psycopg2.connect(**params) con.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT) cur = con.cursor() upd_string_start = f"UPDATE {table_name} SET " upd_string_mid = "" for key in data.keys(): upd_string_mid += f"{key} = %s, " upd_string_end = f"upd_timestamp = %s WHERE name = '{name}'" update_query = upd_string_start + upd_string_mid + upd_string_end list_values = [] for value in data.values(): list_values.append(value) list_values.append(int(time.time())) values = tuple(list_values) cur.execute(update_query, values) cur.close() con.close() print("Data updated successfully.") data = { "population": 619001, "visited": True } if __name__ == "__main__": update_table("Riga", "cities", data)

python update_table.py

Data updated successfully

В pgAdmin мы видим, что в таблице cities снова изменились численность населения Риги , статус посещения и временная отметка.

Удаление БД

Для удаления базы данных используем сетап из предыдущего примера.

Подключаться будем с помощью config.py

Обычно удобно использовать

conn.autocommit = True

Также обратите внимание на то, что имя базы данных заключено в двойные кавычки. Это сделано на случай, если в нём будут дефисы.

import psycopg2 import config def drop_db(db_name: str): print(f"drop_db({db_name})") params = config.credentials() print('Connecting to the PostgreSQL database...') conn = psycopg2.connect(**params) cur = conn.cursor() conn.autocommit = True try: cur.execute(f'DROP DATABASE IF EXISTS "{db_name}"') except (Exception, psycopg2.DatabaseError) as e: print(e) finally: cur.close() conn.close() if __name__ == '__main__': drop_db("obsoleted_db")

Проверить подключение в compose сетапе

Проверить подключение к докер контейнеру с Postgres из другого контейнера

import os import psycopg2 print(os.environ['POSTGRES_USER']) pg_connection_dict = { 'dbname': os.environ['POSTGRES_DB'], 'user': os.environ['POSTGRES_USER'], 'password': os.environ['POSTGRES_PASSWORD'], 'port': 5432, 'host': "postgres_service" } print("pg_connection_dict", pg_connection_dict) con = psycopg2.connect(**pg_connection_dict) print("con", con) sql = "SELECT * FROM pg_catalog.pg_tables;" sql = """SELECT table_schema || '.' || table_name FROM information_schema.tables WHERE table_type = 'BASE TABLE' AND table_schema NOT IN ('pg_catalog', 'information_schema');""" cur = con.cursor() cur.execute(sql) print(cur.fetchall())

Автор статьи: Андрей Олегович