如何使用SQLAlchemy ORM与Cloud Spanner将行读取为模型对象

我正在用FastAPI、SQLAlchemy和Cloud Spanner模拟器开发RESTful API，卡在了把数据库表中一行数据读取为模型类对象的基础操作上。

已定义的模型类 #

class Users(Base):
    __tablename__ = "users"
    __table_args__ = (
        Index("email", "email", unique=True),
        Index("mobile_number", "mobile_number", unique=True),
    )

    id = Column(String(36), primary_key=True)
    first_name = Column(String(20), nullable=False)
    last_name = Column(String(20), nullable=False)
    email = Column(String(50), nullable=False)
    mobile_number = Column(String(13), nullable=False)
    password = Column(String(100), nullable=False)
    role = Column(ForeignKey("roles.id"), nullable=False)
    is_active = Column(Boolean, nullable=False, default=False)
    picture = Column(String(100), nullable=True)

    @hybrid_property
    def name(self):
        """To get the full name of the user."""
        return self.first_name + " " + self.last_name

数据库表中有5条记录，我尝试根据邮箱获取其中一条，但多次尝试均失败：

尝试1：使用filter_by传入表达式 #

代码：

users = session.query(Users).filter_by(Users.email=="kkr@abc.com").all()

错误：

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
TypeError: Query.filter_by() takes 1 positional argument but 2 were given

尝试2：使用filter_by传入字典 #

代码：

users = session.query(Users).filter_by(**{"email":"kkr@abc.com"}).all()

错误：

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/query.py", line 2772, in all
    return self._iter().all()
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/query.py", line 2915, in _iter
    result = self.session.execute(
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 1717, in execute
    result = compile_state_cls.orm_setup_cursor_result(
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/context.py", line 349, in orm_setup_cursor_result
    return loading.instances(result, querycontext)
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/loading.py", line 88, in instances
    with util.safe_reraise():
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/util/langhelpers.py", line 70, in __exit__
    compat.raise_(
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/util/compat.py", line 210, in raise_
    raise exception
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/loading.py", line 69, in instances
    *[
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/loading.py", line 70, in <listcomp>
    query_entity.row_processor(context, cursor)
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/context.py", line 2631, in row_processor
    _instance = loading._instance_processor(
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/loading.py", line 715, in _instance_processor
    primary_key_getter = result._tuple_getter(pk_cols)
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/engine/result.py", line 961, in _tuple_getter
    return self._metadata._row_as_tuple_getter(keys)
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/engine/result.py", line 106, in _row_as_tuple_getter
    indexes = self._indexes_for_keys(keys)
AttributeError: '_NoResultMetaData' object has no attribute '_indexes_for_keys'. Did you mean: '_index_for_key'?

尝试3：改用first() #

代码：

users = session.query(Users).filter_by(**{"email":"kkr@abc.com"}).first()

现象：等待15分钟无响应。

我的需求：只需要将指定邮箱对应的行获取为Users类的对象。

解决方案 #

1. 修正查询语法 #

filter_by接受关键字参数而非表达式，正确写法：

# 方式1：使用filter_by关键字参数
user = session.query(Users).filter_by(email="kkr@abc.com").first()

# 方式2：使用filter支持表达式写法
user = session.query(Users).filter(Users.email == "kkr@abc.com").first()

2. 排查Spanner模拟器连接问题 #

长时间无响应大概率是连接异常，需验证：

• 模拟器是否正常运行（默认端口9010）
• 数据库连接URL配置是否正确，示例：

  engine = create_engine("spanner:///projects/test-project/instances/test-instance/databases/test-db?credentials_path=/dev/null&host=localhost:9010")
  

• 模拟器内users表是否存在，且包含目标邮箱记录
• 网络/防火墙是否阻止了本地连接

3. 验证依赖兼容性 #

确保使用官方适配库google-cloud-spanner-sqlalchemy，且版本与SQLAlchemy、Python版本匹配。

4. 逐步测试排查 #

先执行简单查询验证基础连接：

# 测试连接可用性
result = session.execute("SELECT 1")
print(result.scalar())

# 验证表数据是否可读取
all_users = session.query(Users).all()
print(len(all_users))

内容的提问来源于stack exchange，提问作者Jeet Patel