Python Google Cloud Firestore 错误 504 Deadline Exceeded

Question

我在 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 只是为了查看该函数是否总是在同一个查询中中断，但事实并非如此。

同样在运行该功能后，如果我看到其中一些随机用户，我已经更新了他们的数据，因此它正在运行但在某些时候会中断

Answer 1

persons.stream() 有 60 秒的超时时间。不要在流式传输时处理每个文档，而是尝试预先获取所有文档:

person_docs = [snapshot for snapshot in persons.stream()]

如果您在 60 秒内无法获取更多文档，请尝试使用递归函数 like in this answer 。

订单相同:

orders_docs = [snapshot for snapshot in orders.stream()]