Как стать автором
Поиск
Написать публикацию
Обновить

Комментарии 24

Спасибо за обратную связь)

Рейтинг скрыт

Статья вроде бы неплоха, но есть нюансы, которые в какой-то момент оказываются важны, но не рассмотрены, или просто использованы bad practice

  1. Декоратор def connection(method): – очень не хватает уровня изоляции для транзакции со значением по-умолчанию. Да, в большинстве случаев будет стандартный read commiеted, но когда нужно выбрать другой для всего UnitOfWork – хочется передать его в декоратор, а не писать в коде уродливое await session.execute(text("BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ")), ещё и часто на уровне сервисного слоя, а не уровне репозитория. Ведь всё равно session будет закрыта по завершении вызова декорированного метода

class BaseDAO:
    model = None  # Устанавливается в дочернем классе

Вот этот момент прям боль. И я даже не про отсутствие типа у поля model. У нас же есть дженерики. Почему не использовать их?
Python < 3.12:

import typing

T = typing.TypeVar("T", bound=Base) # мы можем задать границу типа, т.о. мы будем уверены при статическом анализе что использованы верные типы как минимум в иерархии

class BaseDAO(typing.Generic[T]):
    model: type[T]

Python >= 3.12:

# точно так же можно задать границу дженерика 
class BaseDAO[T: Base]:
    model: type[T]

И используем наш T для типизации где хотим:

class BaseDAO[T: Base]:
    model: type[T]

    @classmethod
    async def add(cls, session: AsyncSession, **values: dict[str, typing.Any]) -> T:
        new_instance = cls.model(**values)
        session.add(new_instance)
        try:
            await session.commit()
        except SQLAlchemyError:
            await session.rollback()
            raise
        return new_instance

    @classmethod
    async def add_many(cls, session: AsyncSession, instances: list[dict[str, typing.Any]]) -> list[T]:
        new_instances = [cls.model(**values) for values in instances]
        session.add_all(new_instances)
        try:
            await session.commit()
        except SQLAlchemyError:
            await session.rollback()
            raise
        return new_instances

И сами dao-классы описываются лучше:

class UserDAO(BaseDAO[User]):
    ...


class ProfileDAO(BaseDAO[Profile]):
    ...

Чем же лучше? А тем, что мы не можем не указать класс – статический анализ отвалится. А вот забыть написать model = XXX вполне можно, и mypy даже не ругнётся при model: Base

  1. async def find_all(cls, session: AsyncSession, **filter_by), async def find_one_or_none(cls, session: AsyncSession, **filter_by): и прочие, где передаются **kwargs – я бы сказал а-та-та, очень просто словить из-за опечатки ошибку в запросе из-за того, что передаётся ключ, ссылающийся на несуществующую колонку. Лучше сделать pydantic-модель со всеми нужными опциональными полями, передавать её и делать query = select(сls.model).filter_by(**filters.model_dump(exclude_unset=True))

Рейтинг скрыт

Благодарю за конструктивную критику

Рейтинг скрыт

Попробовал переписать свой pet проект согласно вашему второму замечанию

class BaseORMRepository[T: Base]:
    model: type[T]

class UserManager(BaseORMRepository[User], Singleton):
    @classmethod
    async def create(cls, token_data: TokenMessage, session: AsyncSession):
        user_info = await cls.get_user_info(token_data)
        user = await cls.get("login", user_info.login, session)
        if not user:
            user = cls.model().from_dict(user_info)
            session.add(user)
        cls.__update_token(token=token_data, user=user, session=session)
        await session.commit()

Но получаю ошибку (python 3.12)
:query = Select(cls.model).filter_by(**{arg: value})
^^^^^^^^^
AttributeError: type object 'UserManager' has no attribute 'model'

Рейтинг скрыт

Добрый день!
Да, вы абсолютно правы, я совершенно упустил момент что я у себя использую Dependency Injection, а в данной статье он не рассматривается (я не стал о нём писать, т.к. автор мог либо написать о нём в следующих статьях либо просто его не использовать, все фломастеры разные)
Конечно же в коде, который я написал, у класса BaseDAO и его наследников не будет поля класса (в typing.ClassVar понимании) model, а будет поле экземпляра класса
Судя по тому, что класс UserManager наследуется от Singleton, а session пробрасывается извне, то предположу что у вас user_manager создаётся где-то отдельно и используется заместо того самого DI. В таком случае я бы просто убрал декоратор, сделав методом экземпляра класса, а в методе инициализации сделал user_manager = UserManager(model=User, ...) либо просто переопределил __init__, вызвав super().__init__(model=User)
Склеивая, верный вариант моего кода выше (и наиболее близкий к реальному) выглядит как-то так:

import asyncio
import dataclasses
import typing

import pydantic
import sqlalchemy as sa
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


#  DTO для валидации входных данных
class UserCreateSchema(pydantic.BaseModel):
    first_name: str
    last_name: str


METADATA: typing.Final = sa.MetaData()


class Base(DeclarativeBase):
    metadata = METADATA


class User(Base):
    __tablename__ = "user"

    id: Mapped[typing.Annotated[int, mapped_column(sa.BigInteger, primary_key=True)]]

    first_name: Mapped[typing.Annotated[str, mapped_column(sa.String(length=64))]]
    last_name: Mapped[typing.Annotated[str, mapped_column(sa.String(length=64))]]


@dataclasses.dataclass(kw_only=True)
class BaseORMRepository[T: Base, V: pydantic.BaseModel]:  # T - класс модели-объекта, V - класс модели-валидатора
    model: type[T]

    async def create(self, session: AsyncSession, validated_value: V) -> T:
        user = self.model(**validated_value.model_dump())
        session.add(user)
        # await session.commit() #  закомментировано чтобы код мог запуститься без реального подключения к базе
        return user


class UserManager(BaseORMRepository[User, UserCreateSchema]):
    def __init__(self) -> None:
        super().__init__(model=User)


if __name__ == "__main__":
    user_manager = UserManager()
    session = AsyncSession(bind=None)  # представим что это настоящая сессия
    user_validated = UserCreateSchema(first_name="John", last_name="Doe")
    user_created = asyncio.run(user_manager.create(session, user_validated))
    print(user_created.first_name, user_created.last_name)

Что выведет John Doe в терминал. Да, теряется краткость за счёт переопределения __init__, но приобретается другой момент – все CRUD-методы внутри BaseORMRepository можно правильно типизировать, уменьшив вероятность ошибок

Рейтинг скрыт

Не автор вопроса, но было интересно!

При прочтении кода немного застопорился на моментне:

user = self.model(**validated_value.model_dump())

если заменить user на instance (как пример) будет более очевидно что общий метод.

Рейтинг скрыт

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

Рейтинг скрыт
class BaseManager(Generic[T], Singleton):    
  model: type[T]
  def __init_subclass__(cls):
    cls.model = cls.__orig_bases__[0].__args__[0]

Так тоже будет работать

Рейтинг скрыт

очень полезные статьи. спасибо!

один только вопрос: почему "получим С таблицы", "загрузим С базы" и т. п.? и это не разово, на протяжении 2 статей. вроде бы "ИЗ таблицы". может я чего-то не понимаю?

для новичков полезно было бы добавить echo=True при создании сессии и можно было бы рассматривать какие SQL-запросы алхимия генерирует и выполняет.

Рейтинг скрыт

Спасибо за обратную связь. Возможно просто неграмотно пишу) Казалось что так правильно. По поводу echo планирую в следующей статье описать вывод кастомный под логирование с параметрами. echo перегружает консоль

Рейтинг скрыт

Знаком с алхимией поверхностно, пока нового нашёл мало, но все равно интересно.

Но вот замечание: тема сисек скаляров не раскрыта совсем. В частности не ясно, чем отличается scalar() от scalar_one()

Рейтинг скрыт

В статье блоку скаляров уделена полная глава.

Рейтинг скрыт

Основные методы для работы с результатами запроса

1) scalars()

Метод scalars() используется, когда мы ожидаем получить одну колонку результата, а не несколько полей. Например, когда мы запрашиваем всю модель (как в нашем случае) или одно конкретное поле (например, только имена пользователей).

И далее по тексту)

Рейтинг скрыт

Подскажите, а планируется ли статья по миксинам и hybrid_property / hybrid_method? Первое пригождается когда появляются одинаковые поля (те же id, created_at, deleted и т.д.), а второе позволяет динамически вычислять некие параметры, связанные с объектом БД, без повторений (например посчитать кол-во children при one-to-many и прочее). Если про миксины ещё встречал статьи на хабре, то про hybrid_property как-то не сталкивался, а тема интересная и ИМХО довольно полезная

Рейтинг скрыт

Большое спасибо за статью, а можете подсказать, я только изучаю SQLAlchemy 2, если я классы (например с models.py) - разнесу по файлам (на каждый класс свой файл), при объявлении связей в каждом классе и соответственно импортах - получается цикличный импорт. У меня например есть класс User и Role в отношении многие к одному.
В User:

from models.user import User
users : Mapped[list["User"] | None] = relationship(
        "User",
        back_populates="roles",
        cascade="all, delete-orphan")

В Role:

from models.user import User
users : Mapped[list["User"] | None] = relationship(
        "User",
        back_populates="roles",
        cascade="all, delete-orphan")

Как мне настроить связи таким образом, чтобы роли могли создаваться без указания пользователя при создании? и как решить проблему цикличного импорта?

Рейтинг скрыт

Спасибо за обратную связь. Вы своей реализацией уже решили эту проблему.

Mapped[list["User"] | None]

Проблема была бы при такой записи

Mapped[list[User] | None]

Рейтинг скрыт

Для предотвращения циклических импортов можно юзать TYPE_CHECKING из typing

from typing import TYPE_CHEKING

if TYPE_CHEKING:
  from models.user import User

class Role(Base):
  users: Mapped[list["User"] | None]

Рейтинг скрыт

Интересно. Сталкивались вы с утечкой памяти при использовании алхимии и пидантика? Мы вот недавно были немного в ахтунге, когда приложение раздулось на 5 гигов))

https://github.com/pydantic/pydantic/issues/9429

Рейтинг скрыт

Неплохая вторая часть статьи. Пару замечаний:
1. метод .from_orm считается deprecated, необходимо использовать .model_validate(obj_from_db, from_attributes=True)
2. .dict() также считается deprecated, преобразование pydantic модели в dict лучше производить через .model_dump(), там есть удобные параметры типа by_alias, которые дают гибкость к сериализации данных

Рейтинг скрыт

1. метод .from_orm считается deprecated, необходимо использовать .model_validate(obj_from_db, from_attributes=True)

В статье про Pydantic (самой новой) описывал это.

2. .dict() также считается deprecated, преобразование pydantic модели в dict лучше производить через .model_dump(), там есть удобные параметры типа by_alias, которые дают гибкость к сериализации данных

Та же история. В статье про Pydantic 2 описывал.

Рейтинг скрыт

А почему при двух запросах подряд программа падает c 

Traceback (most recent call last):
  File "C:\Python\lib\asyncio\base_events.py", line 750, in call_soon
    self._check_closed()
  File "C:\Python\lib\asyncio\base_events.py", line 515, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

обязательно все запросы в одной сессии выполнять или что-то в коде не так?

@connection
async def select_all_users(session):
    return await UserDAO.get_all_users(session)
all_users = run(select_all_users())

@connection
async def select_username_id(session):
    return await UserDAO.get_username_id(session)
rez = run(select_username_id())

Рейтинг скрыт

С функциями и сессиями все хорошо, дело в попытке запустить два asyncio.run подряд

Рейтинг скрыт

В вашем случае нужно было под одну сессию вывполнять 2 метода:

@connection
async def select_user(session):
    all_users = await UserDAO.get_all_users(session)
    user_by_id = await UserDAO.get_username_id(session)

Рейтинг скрыт
Зарегистрируйтесь на Хабре, чтобы оставить комментарий

Ваш аккаунт

Разделы

Информация

Услуги

Техническая поддержка
© 2006–2025, Habr