sqlalchemy 异步连接

Question

我正在尝试使用 SQL Alchemy 1.4 和 PostgreSQL 数据库实现基本资源访问。

Python 代码

from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base

Base: DeclarativeMeta = declarative_base()


class User(Base):
  __tablename__ = 'user'

  id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
  email = Column(String(length=255), index=True, nullable=False)


class Resource(Base):
  __tablename__ = "resource"
  
  id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
  name = Column(String(length=255), index=True, nullable=False)


class UserResourceRole(Base):
  __tablename__ = "user_resource_role"

  user_id = Column(
    UUID(as_uuid=True), 
    ForeignKey("user.id", ondelete="CASCADE"),
    primary_key=True
  )
  resource_id = Column(
    UUID(as_uuid=True),
    ForeignKey("resource.id", ondelete="CASCADE"),
    primary_key=True,
  )
  can_edit = Column(Boolean, default=False, nullable=False)

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.ext.asyncio.engine import AsyncEngine
from sqlalchemy.orm import sessionmaker

from the_other_file import User, Resource, UserResourceRole

async def select_all(user: User, db_session: AsyncSession):
  results = await db_session.execute(
    select(Resource, UserResourceRole)
    .join(
      UserResourceRole,
      Resource.id == UserResourceRole.resource_id
    )
    .where(UserResourceRole.user_id == user.id)
  )
  return results.scalars().all()


engine: AsyncEngine = create_async_engine(POSTGRES_URL, future=True)

async_session = sessionmaker(
    bind=engine, class_=AsyncSession, expire_on_commit=False, future=True
)


# ignore the fact that it's not executed in asyncio loop
a_db_session = await async_session()
resources = await select_all(user=a_real_user_is_here, db_session=a_db_session)
print(resources)

我无法在结果中从 UserResourceRole 检索任何内容。它只包含来自Resource 的数据。如果我在select 调用中交换对象，那么我只能从UserResourceRole 检索数据。

我期待什么

我希望得到与该 SQL 查询相同的结果：

SELECT *
FROM resource
INNER JOIN user_resource_role
ON resource.id = user_resource_role.resource_id
WHERE user_resource_role.user_id = :a_user_id

SQL Alchemy 生成的查询完全一样（除了冗长）：

SELECT resource.id, resource.name, user_resource_role.user_id, user_resource_role.resource_id, user_resource_role.can_edit
FROM resource
JOIN user_resource_role
ON resource.id = user_resource_role.resource_id
WHERE user_resource_role.user_id = :user_id_1

Answer 1

如果你尝试

for entry in results:
  print(entry)

它将显示(Resource, UserResourceRole) 的元组列表。显然对.scalars().all() 的调用只留下第一个值。 我目前的解决方案是将结果变成一个列表并手动操作它。