Привет, друзья! Эта статья станет настоящей находкой для всех, кто уже знаком с библиотекой asyncpg-lite, хочет с ней познакомиться или просто стремится легко и эффективно использовать асинхронные возможности SQLAlchemy вместе с asyncpg.

Что такое asynpg-lite?

Представьте себе библиотеку, которая сочетает в себе мощь асинхронного программирования через asyncpg и безграничные возможности SQLAlchemy. Это asynpg-lite — простая и надёжная библиотека, созданная для того, чтобы каждый, даже новичок, мог воспользоваться её потенциалом.

Почему это важно?

В мире, где асинхронное программирование становится всё более востребованным, asynpg-lite предлагает идеальное решение для работы с PostgreSQL. Вам не нужны глубокие знания программирования или сложные концепции. Достаточно базового понимания Python (списки, словари, строки и т.д.) и основ SQL и PostgreSQL (таблицы, CRUD-операции, типы колонок и т.д.).

Что вас ждёт?

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

Готовы погрузиться в мир асинхронных операций на PostgreSQL с помощью asynpg-lite и SQLAlchemy? Тогда давайте начнем!

Начало работы

Для тестирования примеров, которые я приведу, вам потребуется база данных PostgreSQL. Она может быть установлена как на вашем локальном компьютере, так и располагаться удаленно, например, на VPS-сервере.

Как за 5 минут развернуть базу данных PostgreSQL на VPS-сервере, я рассказывал в этой статье: [читать статью].

Сегодня мы разберем все функции библиотеки без разбора кода самой библиотеки (то есть сегодня только практика).

Установка

Для установки последней актуальной версии библиотеки выполните команду:

pip install --upgrade asyncpg-lite

Либо воспользуйтесь requirements.txt (0.3.1.3 самая актуальная версия на данный момент, все версии меньше 0.3 удалены и больше не доступны для скачивания):

asyncpg-lite~=0.3.1.3
pip install -r requirements.txt

Инициация класса

После установки библиотеки необходимо инициировать класс для работы с asynpg-lite, чтобы объект мог подключиться к базе данных и взаимодействовать с ней.

Импортируем библиотеку:

from asyncpg_lite import DatabaseManager

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

• Через DSN-ссылку

• Через данные для подключения (передаете питоновский словарь с данными для подключения).

Вариант инициации через DSN-ссылку

postgresql://USER_LOGIN:PASSWORD@HOST:PORT/DB_NAME

Обратите внимание, сама библиотека будет трансформировать ссылку в корректный формат для взаимодействия с asyncpg (postgresql + postgresql).

Кроме того. Новый формат позволяет указание @ в пароле к подключению.

Пример использования переменных окружения для скрытия DSN-ссылки в файле .env (с использованием библиотеки python-decouple):

from asyncpg_lite import DatabaseManager
from decouple import config
import asyncio

pg_link = config('PG_LINK')
deletion_password = config('ROOT_PASS')


async def main():
    db_manager = DatabaseManager(db_url=pg_link, deletion_password=deletion_password)
    # Далее используем db_manager...

Вариант подключения с указанием данных для подключения

Для этого при инициации объекта менеджера необходимо передать параметр auth_params. Это питоновский словарь, содержащий такие ключи:

• host: str

• port: int

• user: str

• password: str

• database: str

Обязательный параметр: deletion_password

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

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

• expire_on_commit: bool (по умолчанию True)

• echo: bool (по  умолчанию False)

• log_level (по умолчанию logging.INFO)

expire_on_commit

Флаг expire_on_commit используется для управления "устареванием" (expiration) объектов, хранимых в сессии.

expire_on_commit=True (по умолчанию): После вызова commit(), все объекты, которые были изменены в сессии, помечаются как "устаревшие". Это означает, что при следующем обращении к этим объектам они будут автоматически перезагружены из базы данных. Это обеспечивает согласованность данных и гарантирует, что изменения, внесенные другими сессиями или процессами, будут видны.

expire_on_commit=False: Объекты не будут помечены как устаревшие после вызова commit(). Это может быть полезно в случаях, когда вы уверены, что данные в базе данных не изменяются другими процессами и хотите избежать дополнительных запросов к базе данных для повышения производительности.

echo

Флаг echo управляет выводом SQL-запросов в консоль.

echo=False (по умолчанию): Не выводить SQL-запросы в консоль.

echo=True: Выводить все SQL-запросы, выполняемые движком, в консоль. Это полезно для отладки, так как позволяет видеть все запросы, которые отправляются к базе данных.

log_level

log_level – это уровень логирования. Всего бывают следующие варианты:

• DEBUG: Подробная информация, которая в основном полезна для диагностики проблем. Используется для разработки и отладки.

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

• WARNING: Указание на то, что что-то неожиданное произошло, или указывает на проблему в ближайшем будущем (например, «скоро закончится диск»). Программа работает как ожидалось.

• ERROR: Из-за более серьезной проблемы программа не может выполнить некоторую функцию.

• CRITICAL: Серьезная ошибка, из-за которой программа может быть не в состоянии продолжить выполнение.

Общий принцип взаимодействия с менеджером

Для работы с менеджером базы данных из asyncpg_lite необходимо использовать асинхронный контекстный менеджер with. Благодаря такому подходу у вас появляется сразу две возможности:

• Создавать сложные и многоуровневые запросы к базе данных PostgreSQL.

• Не беспокоиться о закрытии соединения с базой данных, так как оно автоматически закрывается после завершения работы с менеджером.

А теперь давайте инициируем наш первый объект класса (я буду использовать словарь для подключения):

from asyncpg_lite import DatabaseManager
import asyncio


from decouple import config
async def main():
    db_manager = DatabaseManager(auth_params=autch_param, deletion_password=root_pass)
    async with db_manager:
        pass

Несмотря на то, что мы не выполнили никаких методов – наши логи покажут:

2024-06-20 21:55:53,233 - INFO - Database instance created with log level: INFO

2024-06-20 21:55:53,301 - INFO - Connect with PostgreSQL success!

2024-06-20 21:55:53,302 - INFO - Disconnected from the database.

На этом примере мы видим, что мы успешно инициировали объект, подключились к базе данных и, так как использовали async with db_manager, а метод запущено не было – соединение автоматически завершилось.

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

Функции библиотеки

Библиотека asyncpg-lite может:

• Создавать таблицы.

• Получать данные (возвращаются в виде списка питоновских словарей или в виде словаря, принимает фильтрацию в виде списков и словарей).

• Добавлять данные в таблицу с обновлением данных при конфликте по первичному ключу или с игнорированием повторного добавления (подробно поясню далее)

• Изменять данные (принимает условие для выбора записей в виде словаря или списка словарей и принимает словарь с новыми данными).

• Удалять данные (может удалять таблицу, все данные из таблицы или значения из таблицы по специальным фильтрам).

Как видите, библиотека покрывает все CRUD операции, используя только базовые типы данных из Python 3.

Создание таблицы

Для создания таблицы используйте метод create_table. Функция принимает:

• table_name: Имя таблицы.

• columns: Список строк, определяющих столбцы таблицы.

Пример кода для создания таблицы пользователей:

async def main():
    from sqlalchemy import BigInteger, String, Integer

    db_manager = DatabaseManager(auth_params=autch_param, deletion_password=root_pass)
    async with db_manager:
        columns = [
            {"name": "user_id", "type": BigInteger, "options": {"primary_key": True, "autoincrement": False}},
            {"name": "user_name", "type": String, "options": {"default": 'Vasya', 'unique': True}},
            {"name": "user_surname", "type": String},
            {"name": "user_age", "type": Integer},
        ] 
await db_manager.create_table(table_name="new_users", columns=columns)

Обратите внимание. Для описания колонок таблиц необходимо использовать список питоновских словарей, где один словарь – это описание одной колонки.

В каждом словаре должно быть 2 обязательных параметра: name (название колонки) и type (тип данных в колонке). Дополнительно можно передать ключ options. Там, к примеру, вы можете указать дефолтное значение, указать что это первичный ключ и прочее.

Единственное что тут может быть непонятным – это описание типов данных. Оно пусть и незначительно, но отличается у SQLAlchemy  от своих аналогов в SQL (PostgreSQL). Ниже приведена сравнительная таблица питоновских типов данных, данных SQL и SQLAlchemy.

Как вы уже поняли каждый тип данных необходимо импортировать из sqlalchemy. Сама sqlalchemy установится автоматически с установкой asyncpg-lite.

Выполним функцию и посмотрим на логи:

2024-06-20 22:27:33,027 - INFO - Database instance created with log level: INFO
2024-06-20 22:27:33,104 - INFO - Connect with PostgreSQL success!
2024-06-20 22:27:35,145 - INFO - Table 'new_users' created with columns: ['user_id', 'user_name', 'user_surname', 'user_age']
2024-06-20 22:27:35,221 - INFO - Disconnected from the database.

Смотрим на таблицу:

Видим, что таблица создана с теми данными что мы хотели (на скрине не видно типа unique, но при повторном добавлении одного и того же имени будет ошибка).

Все остальные методы будут на чистом Python, так что пойдет попроще.

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

Для добавления (обновления) данных в таблицу мы будем использовать метод insert_data_with_update. Данный метод принимает следующие параметры:

• table_name: Имя таблицы, в которую мы будем вставлять данные

• records_data: Словарь или список словарей с данными для вставки.

• conflict_column: Название значения в таблице с PrimaryKey (строка)

• update_on_conflict: Флаг, которым вы укажете нужно ли обновлять данные, если произошел конфликт по первичному ключу. Если поставить True (это по умолчанию), то при конфликте все параметры в строке будут переписаны под тот массив что вы передали. Иначе  конфликтные записи будут просто игнорировать

Метод автоматически определит, передан ли словарь или список словарей.

Попробуем добавить одного пользователя в нашу созданную таблицу:

async def main():
    db_manager = DatabaseManager(auth_params=autch_param, deletion_password=root_pass)
    async with db_manager:
        user_data = {'user_id': 1, 'user_name': 'Вася', 'user_surname': 'Сидоров', 'user_age': 40}
        await db_manager.insert_data_with_update(table_name="new_users",
                                                 records_data=user_data,
                                                 conflict_column='user_id', update_on_conflict=False)

Выполняем и смотри логи:

2024-06-20 23:15:22,570 - INFO - Database instance created with log level: INFO
2024-06-20 23:15:22,639 - INFO - Connect with PostgreSQL success!
2024-06-20 23:15:26,997 - INFO - Inserted/Updated 1 records into table 'new_users'.
2024-06-20 23:15:27,073 - INFO - Disconnected from the database.

Смотрим в таблицу:

Видим, что пользователь добавлен. Теперь попробуем добавить сразу несколько пользователей, с обновлением имени у пользователя с ID 1:

async def main():
    db_manager = DatabaseManager(auth_params=autch_param, deletion_password=root_pass)
    async with db_manager:
        user_data = [
            {'user_id': 1, 'user_name': 'Василий', 'user_surname': 'Сидоренко', 'user_age': 40},
            {'user_id': 2, 'user_name': 'Дмитрий', 'user_surname': 'Иванов', 'user_age': 30},
            {'user_id': 3, 'user_name': 'Олег', 'user_surname': 'Смирнов', 'user_age': 46},
            {'user_id': 4, 'user_name': 'Петр', 'user_surname': 'Петров', 'user_age': 23}
        ]
        await db_manager.insert_data_with_update(table_name="new_users",
                                                 records_data=user_data,
                                                 conflict_column='user_id', update_on_conflict=False)

Смотрим в таблицу:

Видим, что данные не обновились, но, при этом, ошибки мы не получили.

А теперь давайте добавим одного нового пользователя, а у 2 других обновим данные:

async def main():
    db_manager = DatabaseManager(auth_params=autch_param, deletion_password=root_pass)
    async with db_manager:
        user_data = [
            {'user_id': 1, 'user_name': 'Василий', 'user_surname': 'Сидоренко', 'user_age': 40},
            {'user_id': 4, 'user_name': 'Петя', 'user_surname': 'Петров', 'user_age': 24},
            {'user_id': 5, 'user_name': 'Антон', 'user_surname': 'Антонов', 'user_age': 54}
        ]
        await db_manager.insert_data_with_update(table_name="new_users",
                                                 records_data=user_data,
                                                 conflict_column='user_id', update_on_conflict=True)

Выполняем и смотрим в таблицу:

Видим, что данные успешно обновились / добавились. В логах нам вот что сообщают:

2024-06-20 23:21:06,166 - INFO - Database instance created with log level: INFO
2024-06-20 23:21:06,234 - INFO - Connect with PostgreSQL success!
2024-06-20 23:21:10,639 - INFO - Inserted/Updated 3 records into table 'new_users'.
2024-06-20 23:21:10,716 - INFO - Disconnected from the database.

Получение данных

Для получения данных из таблицы используется метод select_data. Метод принимает следующие параметры:

• table_name: Имя таблицы.

• where_dict: Условия для фильтрации данных.

• columns: Список столбцов для извлечения.

• one_dict: Возвращать ли только одну запись в виде словаря.

Получим все значения пользователя с user_id равным 3:

async def main():
    db_manager = DatabaseManager(auth_params=autch_param, deletion_password=root_pass)
    async with db_manager:
        user_info = await db_manager.select_data(table_name="new_users", where_dict={'user_id': 3})
        print(user_info)

Смотрим логи:

2024-06-20 23:23:38,724 - INFO - Database instance created with log level: INFO
2024-06-20 23:23:38,823 - INFO - Connect with PostgreSQL success!

[{'user_id': 3, 'user_name': 'Олег', 'user_surname': 'Смирнов', 'user_age': 46}]

2024-06-20 23:23:43,245 - INFO - Selected 1 rows from table 'new_users'.
2024-06-20 23:23:43,320 - INFO - Disconnected from the database.

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

user_info = await db_manager.select_data(table_name="new_users", where_dict={'user_id': 3}, one_dict=True)

Видим, что возвращен один словарь. Это удобно, когда нужно получить информацию по конкретному пользователю, например, в Telegram-ботах.

Если не передавать аргумент columns, по умолчанию возвращаются все колонки. Можно передать список колонок, значения которых необходимо получить:

user_info = await db_manager.select_data(table_name="new_users", where_dict={'user_id': 3}, one_dict=True, columns=['user_name', 'user_surname'])

Логи:

2024-06-20 23:25:47,402 - INFO - Database instance created with log level: INFO
2024-06-20 23:25:47,471 - INFO - Connect with PostgreSQL success!

{'user_name': 'Олег', 'user_surname': 'Смирнов'}

2024-06-20 23:25:52,167 - INFO - Disconnected from the database.

Теперь получим список всех пользователей со всеми значениями:

users_info = await db_manager.select_data(table_name="new_users")
for i in users_info:
    print(i)

Логи:

2024-06-20 23:27:44,575 - INFO - Database instance created with log level: INFO
2024-06-20 23:27:44,655 - INFO - Connect with PostgreSQL success!

{'user_id': 2, 'user_name': 'Дмитрий', 'user_surname': 'Иванов', 'user_age': 30}
{'user_id': 3, 'user_name': 'Олег', 'user_surname': 'Смирнов', 'user_age': 46}
{'user_id': 1, 'user_name': 'Василий', 'user_surname': 'Сидоренко', 'user_age': 40}
{'user_id': 4, 'user_name': 'Петя', 'user_surname': 'Петров', 'user_age': 24}
{'user_id': 5, 'user_name': 'Антон', 'user_surname': 'Антонов', 'user_age': 54}

2024-06-20 23:27:49,148 - INFO - Selected 5 rows from table 'new_users'.
2024-06-20 23:27:49,226 - INFO - Disconnected from the database.

Теперь усложним запрос и получим значения имени и фамилии у тех пользователей, у которых user_id равен 1, 3 или 4:

user_info = await db_manager.select_data(table_name="new_users",
                                         where_dict=[{'user_id': 1}, {'user_id': 3}, {'user_id': 4}],
                                         columns=['user_surname', 'user_name'])
for user_data in user_info:
    print(user_data)

Логи:

2024-06-20 23:28:55,190 - INFO - Database instance created with log level: INFO
2024-06-20 23:28:55,262 - INFO - Connect with PostgreSQL success!

{'user_surname': 'Смирнов', 'user_name': 'Олег'}
{'user_surname': 'Сидоренко', 'user_name': 'Василий'}
{'user_surname': 'Петров', 'user_name': 'Петя'}

2024-06-20 23:28:59,702 - INFO - Selected 3 rows from table 'new_users'.
2024-06-20 23:28:59,781 - INFO - Disconnected from the database.

Изменение данных

Для изменения данных в библиотеке asyncpg-lite предусмотрен универсальный метод update_data. Данный метод принимает следующие параметры:

• table_name: Имя таблицы.

• where_dict: Условия для выбора записей для обновления (словарь или список словарей).

• update_dict: Словарь с данными для обновления или строка.

Давайте посмотрим на пример использования:

async def main():
    db_manager = DatabaseManager(auth_params=autch_param, deletion_password=root_pass)
    async with db_manager:
        # Пример одиночного словаря
        await db_manager.update_data(
            table_name="new_users",
            where_dict={'user_id': 2},
            update_dict={'user_surname': 'Иванов', 'user_age': 42}
        )

        # Пример списка словарей
        await db_manager.update_data(
            table_name="new_users",
            where_dict=[{'user_id': 2}, {'user_id': 3}],
            update_dict={'user_surname': 'Руссо', 'user_age': 40}
        )


asyncio.run(main())

Логи:

2024-06-20 23:32:37,676 - INFO - Database instance created with log level: INFO
2024-06-20 23:32:37,753 - INFO - Connect with PostgreSQL success!
2024-06-20 23:32:42,026 - INFO - Updated records in table 'new_users' with conditions {'user_id': 2}.
2024-06-20 23:32:43,699 - INFO - Updated records in table 'new_users' with conditions [{'user_id': 2}, {'user_id': 3}].
2024-06-20 23:32:43,860 - INFO - Disconnected from the database.

Обратите внимание: я использовал метод не просто одиночно, а применил серийную обработку данных. На этом примере можно увидеть преимущество использования подхода работы через async with.

Удаление данных

Как я уже описывал выше, в библиотеке предусмотрены три формата удаления данных:

• Удаление записей из таблицы по параметрам (метод delete_data).

• Удаление всех записей из таблицы (метод delete_all_data).

• Удаление таблицы (метод drop_table).

Метод delete_data

Метод delete_data принимает следующие параметры:

• table_name: Имя таблицы.

• where_dict: Условия для выбора записей для удаления (словарь или список словарей).

Давайте удалим из таблицы пользователя с user_id = 1:

await db_manager.delete_data('new_users', {'user_id': 1})

Логи:

2024-06-21 00:08:06,420 - INFO - Database instance created with log level: INFO
2024-06-21 00:08:06,491 - INFO - Connect with PostgreSQL success!
2024-06-21 00:08:10,817 - INFO - Deleted records from table 'new_users' with conditions {'user_id': 1}.
2024-06-21 00:08:10,895 - INFO - Disconnected from the database.

Проверяем:

Видим, что запись удалена.

Теперь удалим пользователей с user_id = 2 и 3:

await db_manager.delete_data('new_users', [{'user_id': 2}, {'user_id': 3}])

Логи:

2024-06-21 00:11:32,750 - INFO - Database instance created with log level: INFO
2024-06-21 00:11:32,847 - INFO - Connect with PostgreSQL success!
2024-06-21 00:11:38,181 - INFO - Deleted records from table 'new_users' with conditions [{'user_id': 2}, {'user_id': 3}].
2024-06-21 00:11:38,277 - INFO - Disconnected from the database.

Смотрим:

Видим, что в таблице остался всего один пользователь. Его мы удалим при помощи метода delete_all_data. Этот метод принимает:

• table_name: Имя таблицы.

• password: Пароль, который должен совпадать с проверочным паролем, указанным при инициализации менеджера.

Код:

await db_manager.delete_all_data('new_users', password=root_pass)

Логи:

2024-06-21 00:15:23,375 - INFO - Database instance created with log level: INFO
2024-06-21 00:15:23,467 - INFO - Connect with PostgreSQL success!
2024-06-21 00:15:28,184 - INFO - Deleted all data from table 'new_users'.
2024-06-21 00:15:28,265 - INFO - Disconnected from the database.

Вся информация успешно удалена.

Метод drop_table принимает те же данные, что и delete_all_data, но удаляет таблицу.

await db_manager.drop_table('new_users', password=root_pass)

Логи:

2024-06-21 00:16:57,158 - INFO - Database instance created with log level: INFO
2024-06-21 00:16:57,228 - INFO - Connect with PostgreSQL success!
2024-06-21 00:17:02,850 - INFO - Table 'new_users' dropped.
2024-06-21 00:17:02,947 - INFO - Disconnected from the database.

Заключение:

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

Ребята, которые имеют большой опыт в разработке и во взаимодействии с PostgreSQL через Python, кто хочет также создать что-то полезное для всех, — выходите со мной на связь любым доступным способом.

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

У меня все. Надеюсь, что проект asyncpg-lite будет развиваться дальше. Спасибо за внимание и жду обратной связи по библиотеке, надеюсь, приятной.

PS. В первом описании к библиотеке, в конце, я давал пратическую задачу. Настоятельно рекомендую ее повторить уже на текущем обновлении.