У меня есть код asyncio sqlalchemy:
import asyncio
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import text, Column, Integer, String
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
async_engine = create_async_engine(f'mysql+aiomysql://root:[email protected]:3306/spy-bot')
AsyncSession = sessionmaker(async_engine, class_=AsyncSession, expire_on_commit=False)
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
username = Column(String(255))
async def main():
async with AsyncSession() as session:
stmt = text('SELECT * FROM `users` LIMIT 1')
result = await session.execute(stmt)
user = result.one()
print(type(user), user)
asyncio.run(main())
как мне сделать экземпляры возврата запроса сеанса в классе User, продолжая использовать необработанный SQL?
В версии синхронизации это будет выглядеть так:
result = session.query(User).from_statement(text(query))
Подробнее здесь: https://stackoverflow.com/questions/755 ... sqlalchemy
Как создать экземпляр модели с использованием необработанного SQL в асинхронной SQLAlchemy? ⇐ Python
-
- Похожие темы
- Ответы
- Просмотры
- Последнее сообщение
-
-
Перехват ошибок из асинхронной функции в блоке try..catch вне асинхронной функции
Anonymous » » в форуме Javascript - 0 Ответы
- 40 Просмотры
-
Последнее сообщение Anonymous
-
-
-
FastApi Sqlalchemy sqlalchemy.exc.InvalidRequestError: не удалось обновить экземпляр
Anonymous » » в форуме Python - 0 Ответы
- 39 Просмотры
-
Последнее сообщение Anonymous
-