【发布时间】:2025-12-07 00:00:02
【问题描述】:
我在对 Google Admin API 进行查询时遇到延迟。
def get_user(self, email: str) -> dict:
res = (
self.service.users()
.list(
domain="gmail.com",
projection="full",
query="email={0}".format(email),
)
.execute()
)
if "users" not in res or len(res["users"]) != 1:
msg = "Could not find user %s" % email
logging.error(msg)
raise GoogleAdminNonExistentUser(msg)
return res["users"][0]
def create_user(
self,
email: str,
first_name: str,
last_name: str,
org_unit_path: str,
manager_email: str,
) -> dict:
user_info = {...}
try:
res = self.service.users().insert(body=user_info).execute()
return res
except HttpError as error:
exc = self._generate_error(error)
logger.exception(exc.message)
raise exc
以这两个调用为例。在我的测试套件中,我进行了创建用户并立即删除它们的测试。在下一个测试中,我创建了相同的用户并更新了自定义属性。然后我验证这些属性是否已设置。
test_create_delete()
create_user(EMAIL)
delete_user(EMAIL)
test_create_update()
create_user(EMAIL) # This will variably error out if the delete_user from the last request hasn't replicated throughout Google
update_user(EMAIL, UPDATE_INFO)
user = get_user()
# This assertion will variably fail if get_user() fetches old data
assert the update info is in user
我可以通过睡眠来增加测试,但构建时间很重要。有没有办法强制 Google Admin API 返回尽可能最新的数据?
【问题讨论】:
-
简单的答案是否定的。这是一项需要时间同步的全球服务,以分钟而不是微秒为单位。
标签: python google-cloud-platform google-admin-sdk