Python3.10+GAE+FastAPI兼容问题：调用NDB服务报错求助

将Python2.7应用迁移至Python3.10，使用FastAPI框架并调用Google App Engine（GAE）遗留捆绑服务时，调用NDB的put()方法触发以下错误：

AssertionError: No api proxy found for service "datastore_v3"

完整错误堆栈：

city.put()
  File "/tmp/tmp7wLPmQ/lib/python3.10/site-packages/google/appengine/ext/ndb/model.py", line 3547, in _put
    return self._put_async(**ctx_options).get_result()
  File "/tmp/tmp7wLPmQ/lib/python3.10/site-packages/google/appengine/ext/ndb/tasklets.py", line 397, in get_result
    self.check_success()
  File "/tmp/tmp7wLPmQ/lib/python3.10/site-packages/google/appengine/ext/ndb/tasklets.py", line 394, in check_success
    six.reraise(self._exception.__class__, self._exception, self._traceback)
  File "/tmp/tmp7wLPmQ/lib/python3.10/site-packages/six.py", line 719, in reraise
    raise value
  File "/tmp/tmp7wLPmQ/lib/python3.10/site-packages/google/appengine/ext/ndb/tasklets.py", line 441, in _help_tasklet_along
    value = gen.throw(exc.__class__, exc, tb)
  File "/tmp/tmp7wLPmQ/lib/python3.10/site-packages/google/appengine/ext/ndb/context.py", line 850, in put
    key = yield self._put_batcher.add(entity, options)
  File "/tmp/tmp7wLPmQ/lib/python3.10/site-packages/google/appengine/ext/ndb/tasklets.py", line 444, in _help_tasklet_along
    value = gen.send(val)
  File "/tmp/tmp7wLPmQ/lib/python3.10/site-packages/google/appengine/ext/ndb/context.py", line 382, in _put_tasklet
    keys = yield self._conn.async_put(options, datastore_entities)
  File "/tmp/tmp7wLPmQ/lib/python3.10/site-packages/google/appengine/datastore/datastore_rpc.py", line 1848, in async_put
    return make_put_call(base_req, pbs, extra_hook)
  File "/tmp/tmp7wLPmQ/lib/python3.10/site-packages/google/appengine/datastore/datastore_rpc.py", line 1828, in make_put_call
    return self._make_rpc_call(config, method, req, resp,
  File "/tmp/tmp7wLPmQ/lib/python3.10/site-packages/google/appengine/datastore/datastore_rpc.py", line 1335, in _make_rpc_call
    rpc = self._create_rpc(config, service_name)
  File "/tmp/tmp7wLPmQ/lib/python3.10/site-packages/google/appengine/datastore/datastore_rpc.py", line 1230, in _create_rpc
    rpc = apiproxy_stub_map.UserRPC(service_name, deadline, callback)
  File "/tmp/tmp7wLPmQ/lib/python3.10/site-packages/google/appengine/api/apiproxy_stub_map.py", line 444, in __init__
    self.__rpc = CreateRPC(service, stubmap)
  File "/tmp/tmp7wLPmQ/lib/python3.10/site-packages/google/appengine/api/apiproxy_stub_map.py", line 69, in CreateRPC
    assert stub, 'No api proxy found for service "%s"' % service
AssertionError: No api proxy found for service "datastore_v3"

用户提供的配置信息：

main.py代码 #

from google.appengine.ext import ndb


class City(ndb.Model):
    """City fields"""
    db_id = ndb.ComputedProperty(lambda self: self.key.id())
    name = ndb.StringProperty(required=True)
    administrative_area_level_1 = ndb.StringProperty(required=True)
    country = ndb.StringProperty(required=True)
    details = ndb.JsonProperty()

dependencies = []

app = FastAPI()


@app.get("/", tags=["root"])
def read_root():
    return {"Imobzi": "Imobzi APP API"}

@app.post("/v1/post")
def set_post():
    city = City(namespace='')
    city.name = 'Teste'
    city.administrative_area_level_1 = "teste"
    city.country = 'testetim'
    city.put()
    return 'test complete'

@app.get("/v1/post")
def get_post():
    print('\n\n', City.query().fetch())
    return {}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8080, reload=True)

app-dev.yaml配置 #

runtime: python310
entrypoint: gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app
app_engine_apis: true

env_variables:

  APPLICATION_ID: dev~None
  GOOGLE_APPLICATION_CREDENTIALS: "./app_engine_service_account_dev.json"
  CLOUDSDK_DEVAPPSERVER_PYTHON: /usr/bin/python2.7

启动脚本bash #

#!/bin/bash

HOST=localhost
PY_PATH=python27=/usr/bin/python2.7,python3=/usr/bin/python3

export APPLICATION_ID=dev~None
export GOOGLE_APPLICATION_CREDENTIALS="./app_engine_service_account_dev.json"
export CLOUDSDK_DEVAPPSERVER_PYTHON=/usr/bin/python2.7

python3 ~/google-cloud-sdk/bin/dev_appserver.py \
    --runtime_python_path=$PY_PATH \
app-dev.yaml

1. 修正GAE API初始化逻辑 #

FastAPI的启动流程和传统GAE Python2环境不同，需要手动初始化GAE API代理和NDB上下文。修改main.py：

import google.appengine.api
from google.appengine.ext import ndb
from google.appengine.api import apiproxy_stub_map
from google.appengine.datastore import datastore_sqlite_stub
from fastapi import FastAPI

# 初始化本地Datastore stub（仅开发环境使用）
if not apiproxy_stub_map.apiproxy.GetStub('datastore_v3'):
    datastore_stub = datastore_sqlite_stub.DatastoreSqliteStub(
        'dev~your-real-project-id',
        '/tmp/dev_appserver.datastore',
        '/tmp/dev_appserver.datastore.indexes'
    )
    apiproxy_stub_map.apiproxy.RegisterStub('datastore_v3', datastore_stub)

# 初始化全局NDB上下文
ndb_context = ndb.Context()
ndb_context.set_cache_policy(lambda key: False)

class City(ndb.Model):
    db_id = ndb.ComputedProperty(lambda self: self.key.id())
    name = ndb.StringProperty(required=True)
    administrative_area_level_1 = ndb.StringProperty(required=True)
    country = ndb.StringProperty(required=True)
    details = ndb.JsonProperty()

app = FastAPI()

@app.get("/", tags=["root"])
def read_root():
    return {"Imobzi": "Imobzi APP API"}

@app.post("/v1/post")
def set_post():
    with ndb_context:
        city = City(namespace='')
        city.name = 'Teste'
        city.administrative_area_level_1 = "teste"
        city.country = 'testetim'
        city.put()
    return 'test complete'

@app.get("/v1/post")
def get_post():
    with ndb_context:
        results = City.query().fetch()
        print('\n\n', results)
    return {}

2. 修正启动配置与环境变量 #

• 移除app-dev.yaml和启动脚本中的CLOUDSDK_DEVAPPSERVER_PYTHON配置，Python3环境无需指定Python2.7路径。
• 将APPLICATION_ID改为真实项目ID格式，比如dev~your-project-id，不要用dev~None。
• 用Python2.7启动dev_appserver.py（GAE开发服务器部分组件依赖Python2），修改启动脚本：

#!/bin/bash

export APPLICATION_ID=dev~your-real-project-id
export GOOGLE_APPLICATION_CREDENTIALS="./app_engine_service_account_dev.json"

python2.7 ~/google-cloud-sdk/bin/dev_appserver.py \
    --runtime_python_path=python3=/usr/bin/python3 \
    app-dev.yaml

3. 改用Python3兼容的NDB库（推荐） #

官方推荐使用google-cloud-ndb替代旧的google.appengine.ext.ndb，该库专为Python3设计，直接调用Cloud Datastore API，无需依赖GAE遗留代理：

1. 安装依赖：

pip install google-cloud-ndb

2. 修改main.py：

from google.cloud import ndb
from fastapi import FastAPI

# 初始化NDB客户端
client = ndb.Client()

class City(ndb.Model):
    db_id = ndb.ComputedProperty(lambda self: self.key.id())
    name = ndb.StringProperty(required=True)
    administrative_area_level_1 = ndb.StringProperty(required=True)
    country = ndb.StringProperty(required=True)
    details = ndb.JsonProperty()

app = FastAPI()

@app.get("/", tags=["root"])
def read_root():
    return {"Imobzi": "Imobzi APP API"}

@app.post("/v1/post")
def set_post():
    with client.context():
        city = City(namespace='')
        city.name = 'Teste'
        city.administrative_area_level_1 = "teste"
        city.country = 'testetim'
        city.put()
    return 'test complete'

@app.get("/v1/post")
def get_post():
    with client.context():
        results = City.query().fetch()
        print('\n\n', results)
    return {}

3. 修改app-dev.yaml，移除app_engine_apis: true：

runtime: python310
entrypoint: gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app

env_variables:
  GOOGLE_APPLICATION_CREDENTIALS: "./app_engine_service_account_dev.json"

内容的提问来源于stack exchange，提问作者Matheus Theu