Учитывая приведенный ниже запрос SQLAlchemy 1.4, я выбираю объект ORM, models.Person
с двумя дополнительными вычисляемыми столбцами, prev
и next
.
@pytest.mark.wip
@pytest.mark.asyncio
async def test_sqlalchemy(people: AsyncSession) -> None:
session = people
query = (
select(
models.Person,
Bundle(
"navigation",
sa.func.lag(models.Person.cursor)
.over(order_by=models.Person.cursor)
.label("prev"),
sa.func.lead(models.Person.cursor)
.over(order_by=models.Person.cursor)
.label("next"),
),
)
.where(models.Person.id > "26744d86-1918-4f67-92a0-e7ca12043721")
.order_by(models.Person.id.asc())
)
iterator = await session.execute(query)
rows = iterator.all()
print(f"person = {rows[0][0]}")
print(f"next = {rows[0].navigation.next} prev = {rows[0].navigation.prev}")
Когда я выполняю запрос и получаю доступ к результирующей строке, я могу получить доступ к следующему и предыдущему столбцам через именованный атрибут, например. rows[0].navigation.next
. Это стало возможным благодаря функции Bundle
в SQLAlchemy.
Однако я могу получить доступ к атрибутам объекта Person
ORM только через индекс столбца, например rows[0][0].surname
. Возможно ли в SQLAlchemy назначить объекту ORM имя столбца, чтобы я мог сделать что-то вроде rows[0].person.surname
?
Я попытался назначить Bundle
объекту ORM, например. Bundle("человек", models.Person). Однако я получаю сообщение об ошибке: AttributeError: 'AnnotatedTable' object has no attribute '_label'
.
Решено на основе информации, предоставленной здесь. Метод заключался в том, чтобы извлечь каждый столбец в переменную, например
person, navigation = rows[0]
print(f"person = {person.surname}")
print(f"next = {navigation.next} prev = {navigation.prev}")
Полное листинговое решение
@pytest.mark.wip
@pytest.mark.asyncio
async def test_sqlalchemy(people: AsyncSession) -> None:
session = people
query = (
select(
models.Person,
Bundle(
"navigation",
sa.func.lag(models.Person.cursor)
.over(order_by=models.Person.cursor)
.label("prev"),
sa.func.lead(models.Person.cursor)
.over(order_by=models.Person.cursor)
.label("next"),
),
)
.where(models.Person.id > "26744d86-1918-4f67-92a0-e7ca12043721")
.order_by(models.Person.id.asc())
)
iterator = await session.execute(query) File not indexed
rows = iterator.all()
person, navigation = rows[0]
print(f"person = {person.surname}")
print(f"next = {navigation.next} prev = {navigation.prev}")