SQLAlchemy 2.0 + Python Generic, или как создать универсальный репозиторий для работы с БД

Доброго времени суток, товарищи, эта статья, так скажем, продолжение предыдущей статьи об SQLAlchemy 2.0 для новичков, в этой статье мы узнаем что такое Python Generic и как его можно использовать в наших целях при взаимодействии с БД.

Первоначальная настройка

Для того чтобы всё правильно работало, необходимо установить MyPy. Это расширение включает более глубокую проверку типов в вашей IDE, для VS Code есть такое расширение от майкрософт.

Установка MyPy в VS Code

Находим расширение, устанавливаем и заходим в параметры VS Code. Там есть следующий пункт: "Mypy-type-checker: Args". Жмём кнопку добавить элемент и вписываем: "mypy-type-checker.args" = ["--enable-incomplete-feature=NewGenericSyntax"]. Это позволяет использовать новый синтаксис объявления дженериков, который был добавлен в Python 3.12.x, обращаю внимание, именно эту версию мы и будем использовать.

Эта же строчка будет информативна при настройке MyPy в других IDE. На этом подготовка заканчивается

Что такое Generic?

Generic, или дженерики - это обобщённый тип данных, для создания обобщённых функций и классов, которые могут работать с любыми типами данных, или только с теми, которые мы укажем, при создании обобщённого типа данных.

Звучит непонятно, поэтому разберём на примере.

Создание обобщённой функции

Создадим обобщённую функцию:

from typing import Type

def foo[T](data: T) -> Type[T]:
	return type(data) # возвращаем тип аргумента
	
print(foo(5),
	  foo("hello"),
	  foo(4.56)
)

Вывод в терминале:

<class 'int'> <class 'str'> <class 'float'>

[T] - создаёт наш обобщённый тип данных T, который в будущем, при вызове функции станет любым типом данных, объект которого вы передадите в аргументе.
А теперь попробуйте заменить [T] на [T: (int, float)] и ваша IDE начнёт жаловаться на foo("hello"), потому что вы ограничили обобщённый тип, однако, если вы запустите код, то всё будет работать.

def foo[T: (int, float)](data: T) -> T:
	return type(data) # возвращаем тип аргумента
	
print(foo(5),
	  foo("hello"), # не ОК
	  foo(4.56)
)

Вы, наверное, думаете: "Я же это мог делать и без дженериков, да и реальных ограничений они не создают, так зачем они мне тогда нужны?". На самом деле всё просто - это добавит условно строгую типизацию в ваш код, но только на уровне анализа кода(как раз с помощью MyPy), самому Python без разницы на ваши указания, динамическая типизация есть динамическая типизация. И это, отчасти, удобно, но создаёт трудности в читаемости кода. Если вы решите не использовать дженерики для обобщённых функций/классов, то попрощайтейсь с подсказками от вашей IDE и читаемостью кода.

Создание обобщённого класса

Теперь напишем более сложную структуру - обобщённый класс:

class Stack[T]:
    def __init__(self) -> None:
        # Создание пустого списка с типом данных T
        self.items: list[T] = []

    def push(self, item: T) -> None:
        self.items.append(item)

    def pop(self) -> T:
        return self.items.pop()

    def empty(self) -> bool:
        return not self.items

stack = Stack[int]()
stack.push(5)

При создании объекта класса и вызове функции push(), наблюдаем подсказку следующего вида:

Если мы попробуем добавить число типа float, то увидим следующее:

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

Создание универсального репозитория с помощью Generic и SQLALchemy

Для начала добавим необходимые импорты:

from typing import Type, Sequence

from sqlalchemy import select 
from sqlalchemy.orm import sessionmaker 

from app.repositories.database import engine

Первый импорт: Type понадобится при передаче модели данных вовнутрь репозитория, Sequence - тип данных "Последовательность", нужен, т.к. функция all() возвращает данные такого типа.

Второй и третий импорты: функция select для получения данных из БД, класс sessionmaker для создания сессии

Четвёртый импортирует интерфейс для взаимодействия с БД, который мы создавали ранее в предыдущей статье (если не знаете что это такое - читайте предыдущую статью, в контексте данной темы информация о том, как подключать БД будет излишней).

Затем мы создаём класс Repository, который будет взаимодействовать с БД:

class Repository[M]:
	def __init__(self, model: Type[M]) -> None:
		self.Model: Type[M] = model

model: Type[M] - через аргумент функции-конструктора передаём модель данных вовнутрь репозитория. Type в данной ситуации нужен, т.к. мы передаём именно класс, а не его экземпляр.

Представим, что у нас есть две модели данных: Category и Product, создадим для них репозитории:

# добавляем соответствующие импорты
from app.models.category import Category
from app.models.product import Product

...

category_repo = Repository[Category](Category)
product_repo = Repository[Product](Product)

Теперь объясняю: в квадратных скобках [Category] мы передаём тип данных, т.е. благодаря нему у нас появятся подсказки при использовании функций, в обычных скобках (Category) мы передаём модель данных, чтобы мы могли внутри функций работать с экземплярами модели данных.

Напишем функцию для создания объектов в БД:

def create(self, new_object: M, session) -> None:
	session.add(new_object)

Обратите внимание на new_object: M, здесь мы добавляем аннотацию типа, как это сыграет нам на руку? Вот так:

А ведь при объявлении функции мы написали M, а не Category, но наша IDE понимает как работают дженерики и даёт нам правильные подсказки.
P.S. красным подсвечивается из-за пустых скобок.

Теперь напишем функцию для получения всех объектов из БД:

def get_all(self) -> Sequence[M]:
	statement = select(self.Model)
	objects = session.scalars(statement).all()
	return objects

Взглянем на функцию get_all(): внутри мы видим объявленную нами self.Model, которая является переданной нами при инициализации экземпляра репозитория моделью данных, и теперь она может использоваться для таких функций, как select, а также при использовании get_all() мы можем наблюдать следующее:

Т.е. IDE показывает тип данных, которые функция нам вернёт.

Весь код:

from typing import Type, Sequence

from sqlalchemy import select
from sqlalchemy.orm import sessionmaker

from app.repositories.database import engine
from app.models.category import Category
from app.models.product import Product

Session = sessionmaker(engine)

class Repository[M]:
	def __init__(self, model: Type[M]) -> None:
		self.Model = model

	def create(self, new_object: M, session) -> None:
		session.add(new_object)
		session.commit()

	def get_all(self, session) -> Sequence[M]:
		statement = select(self.Model)
		objects = session.scalars(statement).all()
		return objects

category_repo = Repository[Category](Category)
product_repo = Repository[Product](Product)

electronics = Category(name="Электроника")

# пример использования
with Session() as session:
  try:
      print(category_repo.get_all(session))
      category_repo.create(electronics, session)
  except:
      session.rollback()
      raise
  else:
      session.commit()
      session.refresh()
      print(category_repo.get_all(session))

Лучший способ запомнить информацию - применить на практике, поэтому я предлагаю Вам создать функции update(), get_by_name() и delete(). Варианты этих функций без дженериков имеются в предыдущей статье.

Подведём итог: дженерики добавляет условно-строгую типизацию для нас, других разработчиков и IDE, но не для интерпретатора Python! Что позволяет сделать наш код более читабельным, а также добавляет подсказки типов от IDE.

Если вы хотите сделать код максимально читабельным, можете создавать репозиторий под каждую модель данных, но если у Вас таковых штук 50... выводы делайте сами :)