在 SqlAlchemy 的异步 orm 引擎中,如何查询一个表并获取一个值或全部?
我知道对于非异步方法,我只需要执行:
SESSION.query(TableClass).get(x)
但是在尝试使用异步方法时,它会抛出以下错误:
AttributeError: 'AsyncSession' object has no attribute 'query'.
下面是我定义的 SESSION 变量。LOOP 变量仅为 asyncio.get_event_loop()
,用于在加载我的 sql 模块并填充用作缓存的变量以避免每次需要某些内容时都进行数据库缓存时启动异步方法:
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session
from .. import CONFIG, LOOP
def _build_async_db_uri(uri):
if "+asyncpg" not in uri:
return '+asyncpg:'.join(uri.split(":", 1))
return uri
async def start() -> declarative_base:
engine = create_async_engine(_build_async_db_uri(CONFIG.general.sqlalchemy_db_uri))
async with engine.begin() as conn:
BASE.metadata.bind = engine
await conn.run_sync(BASE.metadata.create_all)
return scoped_session(sessionmaker(bind=engine, autoflush=False, class_=AsyncSession))
BASE = declarative_base()
SESSION = LOOP.run_until_complete(start())
这里是一个表格和缓存函数的示例:
class TableClass:
__tablename__ = "tableclass"
id = Column(Integer, primary_key = True)
alias = Column(Integer)
CACHE = {}
async def _load_all():
global CACHE
try:
curr = await SESSION.query(TableClass).all()
CACHE = {i.id: i.alias for i in curr}
LOOP.run_until_complete(_load_all())
q = select(TableClass)
的确切名称,搜索起来会更容易。 - Harshit Singhai