Комментарии 16
@classmethod
async def find_all(cls):
async with async_session_maker() as session:
query = select(cls.model)
result = await session.execute(query)
return result.scalars().all()Автору прежде чем писать гайды, надо почитать документацию, почему не надо создавать сессию в самом DAO
https://docs.sqlalchemy.org/en/20/orm/session_basics.html
def __str__(self):
return f"{self.__class__.__name__}(id={self.id}, major_name={self.major_name!r})"
def __repr__(self):
return str(self)Ужас... Можно же просто переопределить метод repr
Пожалуйста, указывайте лучший код с комментариями. Чтоб было всем понятно о чем речь. Благодарю
def __repr__(self):
return f"{self.__class__.__name__}(id={self.id}, major_name={self.major_name!r})"Все, больше ничего не надо делать. Если метод dunder str не определен, то питон автоматом вызывает dunder repr метод.
А прежде чем писать такие статьи, стоит почитать документацию питона. Вы же понимаете что многие люди будут ваши статьи, особенно новички, потом нахватаются от вас плохого кода. Благодарю
@classmethod
async def find_all(cls):
async with async_session_maker() as session:
query = select(cls.model)
result = await session.execute(query)
return result.scalars().all()Хорошо, что вы попробовали и не побоялись выложить, учитывая какие софт-скиллы у многих комментаторов на Хабре. Можно сделать лучше. Приведу примеры для синхронной и асинхронной алхимии, как сделать лучше
ПОРАБОТАЕМ В СИНХРОННОЙ АЛХИМИИ
В обычной синхронной алхимии используется, как правило, либо execute для выполнения произвольных SQL-запросов и возвращает объект ResultProxy, либо scalars для выполнения запросов, которые возвращают ровно один столбец и одну строку, и возвращает значение этой единственной ячейки напрямую. То есть если бы это была обычная синхронная сессия, такой синтаксис был бы далеко не самым лучшим решиением.
Код для синхронной алхимии:
# импортируем то, что нужно
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# где-то ранее создаем движок и сессию
engine = create_engine(os.environ.get('DATABASE_URL'))
Session = sessionmaker(engine)
# теперь работаем в классе
@classmethod
def find_all(cls):
with Session() as session:
query = select(cls.model)
return session.scalars(query)У scalars не нужно писать all() - оно по умолчанию установлено
ТЕПЕРЬ ПОГОВОРИМ ОБ АСИНХРОННОЙ АЛХИМИИ
Как верно было замечено выше надо создать отдельную асинхронную сессию. У движка дожен быть асинхронный драйвер, например, asyncpg для PostgreSQL
# импортируем все необходимое для работы
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
# создаем движок и сессию
engine = create_async_engine(os.environ.get('DATABASE_URL'))
Session = async_sessionmaker(engine, expire_on_commit=False)
# теперь работаем в классе
@classmethod
async def find_all(cls):
async with Session() as session:
query = select(cls.model)
return await session.stream_scalars(query)Метод session.stream_scalars(query) обеспечивают возврат асинхронной версии объекта, который поддерживает протокол асинхронной итерации Python.То есть для его итерации вы будете использовать не простой цикл for, а асинхронный async for . Он позволяет выполнять асинхронные операции в процессе итерации. Итерации выполняются в асинхронном контексте, что позволяет эффективно использовать время CPU при ожидании завершения асинхронных операций вместо блокировки процесса.
И на последок небольшую психологическую поддержку Вам хочу оказать: продолжайте дальше писать статьи, вы работаете с довольно сложными вещами. И официальный туториал по SQLAlchemy написан, слабо говоря, не лучшим образом. Книга только одна вменяемая по этой теме - вот эта. Ну а к хамству и грубости надо просто привыкнуть - как сказал не последний человек в мире Python Никита Соболев, что в России все отлично с хардами, но все ужасно с софтами. Грубость в комментариях к вам никакого отношения не имеет, этих людей так с детсва воспитало наше общество
Благодарю за обратную связь. Правда. Очень радует, что есть такие комментаторы как вы. Благодарю и за примеры качественного кода. Обязательно внедрю его в будущие статьи.
У scalars не нужно писать all() - оно по умолчанию установлено
Это не так. result.scalars() привязан к соединению из пула и к сессии, result.scalars().all() нет. В теории, это может вызвать проблемы при формировании ответа на запрос, ведь сессия в данный момент уже будет закрыта.
Метод session.stream_scalars(query) обеспечивают возврат асинхронной версии объекта, который поддерживает протокол асинхронной итерации Python.
stream_scalars по большей части нужен для работы с большим объёмом данных. Если данных немного, то лучше сразу забрать их все через scalars().all().
То есть для его итерации вы будете использовать не простой цикл for, а асинхронный async for . Он позволяет выполнять асинхронные операции в процессе итерации.
async for служит для других целей. С синхронным итератором вы тоже можете выполнять асинхронные операции в процессе итерации.
У вас в разных pydantic классах используется синтаксис версий 1 и 2 (декораторы validator и field_validator). Тут либо одно, либо другое.
Спасибо! пошёл читать 5 часть +)
Уточню, при создании новой записи в БД обычно принято возвращать созданный объект?
То есть создаю нового студента и в случае успеха получаю объект студент из БД с присвоенным PK.
Здесь у вас возвращаем сообщение с обектом, который подавали на вход. Это для упрощения?
Интересно было бы увидеть подключение tokena в swagger
https://habr.com/ru/articles/829742/
В этой статье описывал подробно
http://127.0.0.1:8000/students/by_filter?enrollment_year=2008и вот такая фигня
Error: Unprocessable Entity
Response body
{
"detail": [
{
"type": "missing",
"loc": [
"query",
"student_id"
],
"msg": "Field required",
"input": null
}
]
}RBStudent из примера.
def __init__(self, student_id: int | None = None,почему оно хочет обязательного указания student_id?
Создание собственного API на Python (FastAPI): Router и асинхронные запросы в PostgreSQL (SQLAlchemy)