How do I poll Google Long-Running Operations using Python Library?

user_7685955 0
Viewed 155

I have a google.api_core.operation.Operation object from

operation = client.async_batch_annotate_files(requests=[async_request])

I would like to check the status of that operation. I'm trying to use google.api_core.operations_v1.AbstractOperationsClient. It has the method, get_operation, which should return the status.

client=AbstractOperationsClient()
res=client.get_operation(operation.operation.name)

I get an error when that is run:

ValueError: Request {'name': '*redacted*'} does not match any URL path template in available HttpRule's ['/v1/{name=operations/**}']

The error is generated with this code, I believe.

google-cloud-platform python
1 Answers

Something I could get working is using the OperationsClient that is implemented and exposed by the Vision API ImageAnnotatorClient (similar to this answer in another thread):

ops_client = vision_v1.ImageAnnotatorClient().transport.operations_client

Using this client works when passing the full operation name, which was giving you the initial error about URL templates:

ops_client.get_operation(vision_operation.operation.name)

Querying get_operation() gives you the expected metadata of the operation, such as the done status to detect when the operation finishes:

ops_client.get_operation(vision_operation.operation.name).done
# Output
False

Otherwise, if you need a simpler way of querying status, the running() method of the Vision API Operation object returns a simple True or False depending on completion status. You don't have to instantiate an OperationsClient this way.

vision_operation.running()

Full snippet (Based off this guide):

def main():
    input_image_uri="gs://cloud-samples-data/vision/label/wakeupcat.jpg"
    output_uri=<destination_bucket>

    ops_client = vision_v1.ImageAnnotatorClient().transport.operations_client

    # get_vision_op() not included for simplicity, returns the Operation out of async_batch_annotate_images() as shown on the linked guide
    vision_operation = get_vision_op(input_image_uri, output_uri)

    print(ops_client.get_operation(vision_operation.operation.name).done)
    print(vision_operation.running())

    print("Waiting for operation to complete...")
    response = vision_operation.result(90)

    print(vision_operation.running())
    print(ops_client.get_operation(vision_operation.operation.name).done)
    # Output sent to destination bucket
    gcs_output_uri = response.output_config.gcs_destination.uri
    print("Output written to GCS")
user_169298940
Related