我在 python 中有一个函数 for firestore,我为一个集合的所有用户执行 for 循环,然后我进入另一个集合以获取一些指标,并在第一个集合中更新这些指标。
我运行了该函数,但在执行过程中的某个时刻,该函数中断了,给了我这个错误:
_Rendezvous Traceback (most recent call last)
~\Anaconda3\envs\work\lib\site-packages\google\api_core\grpc_helpers.py in next(self)
78 try:
---> 79 return six.next(self._wrapped)
80 except grpc.RpcError as exc:
~\Anaconda3\envs\work\lib\site-packages\grpc\_channel.py in __next__(self)
363 def __next__(self):
--> 364 return self._next()
365
~\Anaconda3\envs\work\lib\site-packages\grpc\_channel.py in _next(self)
346 else:
--> 347 raise self
348 while True:
_Rendezvous: <_Rendezvous of RPC that terminated with:
status = StatusCode.DEADLINE_EXCEEDED
details = "Deadline Exceeded"
debug_error_string = "{"created":"@1570660422.708000000","description":"Error received from peer ipv4:216.58.202.234:443","file":"src/core/lib/surface/call.cc","file_line":1052,"grpc_message":"Deadline Exceeded","grpc_status":4}"
>
The above exception was the direct cause of the following exception:
DeadlineExceeded Traceback (most recent call last)
<ipython-input-20-05c9cefdafb4> in <module>
----> 1 update_collection__persons()
<ipython-input-19-6e2bdd597a6e> in update_collection__persons()
10 counter_secs = 0
11
---> 12 for person_doc in person_docs:
13 person_dict = person_doc.to_dict()
14 last_updated = person_dict['last_updated']
~\Anaconda3\envs\work\lib\site-packages\google\cloud\firestore_v1\query.py in stream(self, transaction)
766 )
767
--> 768 for response in response_iterator:
769 if self._all_descendants:
770 snapshot = _collection_group_query_response_to_snapshot(
~\Anaconda3\envs\work\lib\site-packages\google\api_core\grpc_helpers.py in next(self)
79 return six.next(self._wrapped)
80 except grpc.RpcError as exc:
---> 81 six.raise_from(exceptions.from_grpc_error(exc), exc)
82
83 # Alias needed for Python 2/3 support.
~\Anaconda3\envs\work\lib\site-packages\six.py in raise_from(value, from_value)
DeadlineExceeded: 504 Deadline Exceeded
一直在找解决办法,资料不多,这里发现了一个类似的问题:https://github.com/googleapis/google-cloud-python/issues/8933
所以我尝试使用此代码但不起作用。这是我的功能:
def update_collection__persons():
persons = db.collection(u'collections__persons')
person_docs = persons.stream()
counter_secs = 0
for person_doc in person_docs:
person_dict = person_doc.to_dict()
last_updated = person_dict['last_updated']
last_processed = person_dict['last_processed']
dt_last_updated = datetime(1, 1, 1) + timedelta(microseconds=last_updated/10)
dt_last_processed = datetime(1, 1, 1) + timedelta(microseconds=last_processed/10)
if dt_last_processed < dt_last_updated:
orders = db.collection(u'collection__orders').where(u'email', u'==', person_dict['email'])
orders_docs = orders.stream()
sum_price = 0
count = 0
date_add_list = []
for order_doc in orders_docs:
order_dict = order_doc.to_dict()
sum_price += order_dict['total_price']
count +=1
date_add_list.append(order_dict['dateAdded'])
if count > 0:
data = {'metrics': {'LTV': sum_price,
'AOV': sum_price/count,
'Quantity_orders': count,
'first_order_date': min(date_add_list),
'last_order_date': max(date_add_list)},
'last_processed': int((datetime.utcnow() - datetime(1, 1, 1)).total_seconds() * 10000000)}
db.collection(u'collection__persons').document(person_dict['email']).set(data, merge = True)
我创建了一个 counter_secs 只是为了查看该函数是否总是在同一个查询中中断,但事实并非如此。
同样在运行该功能后,如果我看到其中一些随机用户,我已经更新了他们的数据,因此它正在运行但在某些时候会中断
最佳答案
persons.stream()
有 60 秒的超时时间。不要在流式传输时处理每个文档,而是尝试预先获取所有文档:
person_docs = [snapshot for snapshot in persons.stream()]
如果您在 60 秒内无法获取更多文档,请尝试使用递归函数 like in this answer 。
订单相同:
orders_docs = [snapshot for snapshot in orders.stream()]
关于Python Google Cloud Firestore 错误 504 Deadline Exceeded,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/58313177/