Alex | maked0n1an
Привет, ответьте на мои вопросы, пожалуйста, для прояснения картины. Каша уже в голове, несколько часов потратил.
Как использовать UoW (https://t.me/advice17/60) от SQLAlchemy ORM в след. структуре (роутер -> сервис/бизнес логика -> слой данных к бд для сокрытия деталей реализации). Перечитал все посты, связанные с алхимией в "Советы разработчикам (python и не только)", прочитал все вопросы и комментарии, что задавали до меня здесь.
Итак, SQLAlchemy реализует паттерн Data Mapper (https://t.me/advice17/66, https://techspot.zzzeek.org/2012/02/07/patterns-implemented-by-sqlalchemy/), SQLAlchemy.Session | AsyncSession под капотом реализует UoW и неявно Repository (с ласт ссылки):
Есть много противоречий по поводу того, создавать ли "репозиторий" поверх всего этого или нет, но в этом посте (https://t.me/advice17/74) сказано, что это имеет место быть:
1) Получается, мы создаем обертку в виде репозитория для управления классом, унаследованного от sqlalchemy.DeclarativeBase. Это избыточно же, нет?
2) Верно ли это утверждение?
3) UoW инжектируется в класс сервиса (https://t.me/sqlalchemy_help/123336). Насколько я правильно понял, это вновь созданная нами избыточная обертка вокруг sqlalchemy.Session, без которой никак не обойтись, чтобы в одной транзакции можно было манипулировать данные со связанных таблиц, верно?
Вот мой репозиторий с учетом того, что было написано ранее: https://github.com/maked0n1an/api-currency-converter/blob/main/src/services/user.py
Я так понимаю, что здесь создание UnitOfWork, что управляет Repos, является неправильным и есть какой-то другой способ.
4) Последний вопрос: есть реализации,
4.1) когда сессия создается прямо в роутере через Depends() и передается через все слои: роутер -> бизнес-логика -> слой для работы с бд (https://github.com/maked0n1an/api-currency-converter/blob/main/src/api/dependencies/dependencies.py#L27);
4.2) создается при инициализации созданного нами UnitOfWork и тот прокидывается тоже через dependency (https://github.com/artemonsh/fastapi-onion-architecture/blob/pt2_unit_of_work/src/utils/unitofwork.py#L37):
class UnitOfWork:
def __init__(self):
self.session_factory = async_session_maker
async def __aenter__(self):
self.session = self.session_factory()
self.users = UsersRepository(self.session)
self.tasks = TasksRepository(self.session)
self.task_history = TaskHistoryRepository(self.session)
4.3) сессия создается в хендлере @db_session_handler для методов слоя данных к бд и добавление sqlalchemy.Session через kwargs:
def db_session_handler(coroutine: Callable[..., Awaitable[Any]]):
async def wrapper(*args, **kwargs):
if session := kwargs.pop("session", None):
assert isinstance(session, AsyncSession)
kwargs["session"] = session
result = await coroutine(*args, **kwargs)
else:
async with Services.database.session() as session:
kwargs["session"] = session
result = await coroutine(*args, **kwargs)
return result
return wrapper
Что из них лучше всего?
Как использовать UoW (https://t.me/advice17/60) от SQLAlchemy ORM в след. структуре (роутер -> сервис/бизнес логика -> слой данных к бд для сокрытия деталей реализации). Перечитал все посты, связанные с алхимией в "Советы разработчикам (python и не только)", прочитал все вопросы и комментарии, что задавали до меня здесь.
Итак, SQLAlchemy реализует паттерн Data Mapper (https://t.me/advice17/66, https://techspot.zzzeek.org/2012/02/07/patterns-implemented-by-sqlalchemy/), SQLAlchemy.Session | AsyncSession под капотом реализует UoW и неявно Repository (с ласт ссылки):
Unit of Work - This pattern is where the system transparently keeps track of changes to objects and periodically flushes all those pending changes out to the database. SQLAlchemy's Session implements this pattern fully in a manner similar to that of Hibernate.
Repository - An interface that serves as the gateway to the database, in terms of object-relational mappings. This is the SQLAlchemy Session.
Есть много противоречий по поводу того, создавать ли "репозиторий" поверх всего этого или нет, но в этом посте (https://t.me/advice17/74) сказано, что это имеет место быть:
Даже используя ORM приходится писать сложные select запросы для реализации поиска или получения сводной информации, поэтому имеет смысл реализовать паттерн репозиторий, скрывая их построение внутри, но при этом мы все ещё полагаемся на возможности сессии как UoW.
1) Получается, мы создаем обертку в виде репозитория для управления классом, унаследованного от sqlalchemy.DeclarativeBase. Это избыточно же, нет?
2) Верно ли это утверждение?
В случае создания репозитория объект sqlalchemy.Session передается в метод init репозитория.
3) UoW инжектируется в класс сервиса (https://t.me/sqlalchemy_help/123336). Насколько я правильно понял, это вновь созданная нами избыточная обертка вокруг sqlalchemy.Session, без которой никак не обойтись, чтобы в одной транзакции можно было манипулировать данные со связанных таблиц, верно?
Вот мой репозиторий с учетом того, что было написано ранее: https://github.com/maked0n1an/api-currency-converter/blob/main/src/services/user.py
Я так понимаю, что здесь создание UnitOfWork, что управляет Repos, является неправильным и есть какой-то другой способ.
4) Последний вопрос: есть реализации,
4.1) когда сессия создается прямо в роутере через Depends() и передается через все слои: роутер -> бизнес-логика -> слой для работы с бд (https://github.com/maked0n1an/api-currency-converter/blob/main/src/api/dependencies/dependencies.py#L27);
4.2) создается при инициализации созданного нами UnitOfWork и тот прокидывается тоже через dependency (https://github.com/artemonsh/fastapi-onion-architecture/blob/pt2_unit_of_work/src/utils/unitofwork.py#L37):
class UnitOfWork:
def __init__(self):
self.session_factory = async_session_maker
async def __aenter__(self):
self.session = self.session_factory()
self.users = UsersRepository(self.session)
self.tasks = TasksRepository(self.session)
self.task_history = TaskHistoryRepository(self.session)
4.3) сессия создается в хендлере @db_session_handler для методов слоя данных к бд и добавление sqlalchemy.Session через kwargs:
def db_session_handler(coroutine: Callable[..., Awaitable[Any]]):
async def wrapper(*args, **kwargs):
if session := kwargs.pop("session", None):
assert isinstance(session, AsyncSession)
kwargs["session"] = session
result = await coroutine(*args, **kwargs)
else:
async with Services.database.session() as session:
kwargs["session"] = session
result = await coroutine(*args, **kwargs)
return result
return wrapper
Что из них лучше всего?