Google Cloud Vision Operators¶
Prerequisite Tasks¶
To use these operators, you must do a few things:
Select or create a Cloud Platform project using the Cloud Console.
Enable billing for your project, as described in the Google Cloud documentation.
Enable the API, as described in the Cloud Console documentation.
Install API libraries via pip.
pip install 'apache-airflow[google]'Detailed information is available for Installation.
CloudVisionAddProductToProductSetOperator¶
Creates a new ReferenceImage
resource.
For parameter definition, take a look at
CloudVisionAddProductToProductSetOperator
Using the operator¶
We are using the Product
,
ProductSet
and Retry
objects from
Google libraries:
from google.api_core.retry import Retry # isort:skip
from google.cloud.vision_v1.types import ProductSet # isort:skip
from google.cloud.vision_v1.types import Product # isort:skip
If product_set_id
and product_id
was generated by the API it can be extracted from XCOM:
add_product_to_product_set = CloudVisionAddProductToProductSetOperator(
location=LOCATION,
product_set_id=product_set_create_output,
product_id="{{ task_instance.xcom_pull('product_create') }}",
retry=Retry(maximum=10.0),
timeout=5,
task_id="add_product_to_product_set",
)
Otherwise it can be specified explicitly:
add_product_to_product_set_2 = CloudVisionAddProductToProductSetOperator(
location=LOCATION,
product_set_id=GCP_VISION_PRODUCT_SET_ID,
product_id=GCP_VISION_PRODUCT_ID,
retry=Retry(maximum=10.0),
timeout=5,
task_id="add_product_to_product_set_2",
)
Templating¶
template_fields: Sequence[str] = (
"location",
"product_set_id",
"product_id",
"project_id",
"gcp_conn_id",
"impersonation_chain",
)
More information¶
See Google Cloud Vision Add Product To Product Set documentation.
CloudVisionImageAnnotateOperator¶
Run image detection and annotation for an image.
For parameter definition, take a look at
CloudVisionImageAnnotateOperator
Using the operator¶
We are using the enums
and Retry
objects from
Google libraries:
from google.api_core.retry import Retry # isort:skip
from google.cloud.vision_v1 import Feature # isort:skip
annotate_image = CloudVisionImageAnnotateOperator(
request=annotate_image_request,
retry=Retry(maximum=10.0),
timeout=5,
task_id="annotate_image",
)
The result can be extracted from XCOM:
annotate_image_result = BashOperator(
bash_command="echo {{ task_instance.xcom_pull('annotate_image')"
"['logoAnnotations'][0]['description'] }}",
task_id="annotate_image_result",
)
Templating¶
template_fields: Sequence[str] = (
"request",
"gcp_conn_id",
"impersonation_chain",
)
More information¶
CloudVisionCreateProductOperator¶
Creates and returns a new product resource.
Possible errors regarding the Product
object provided:
Returns INVALID_ARGUMENT if
display_name
is missing or longer than 4096 characters.Returns INVALID_ARGUMENT if
description
is longer than 4096 characters.Returns INVALID_ARGUMENT if
product_category
is missing or invalid.
For parameter definition, take a look at
CloudVisionCreateProductOperator
Using the operator¶
We are using the Product
and Retry
objects from Google libraries:
from google.cloud.vision_v1.types import Product # isort:skip
from google.api_core.retry import Retry # isort:skip
product = Product(display_name="My Product 1", product_category="toys")
The product_id
argument can be omitted (it will be generated by the API):
product_create = CloudVisionCreateProductOperator(
location=LOCATION,
product=product,
retry=Retry(maximum=10.0),
timeout=5,
task_id="product_create",
)
Or it can be specified explicitly:
product_create_2 = CloudVisionCreateProductOperator(
product_id=GCP_VISION_PRODUCT_ID,
location=LOCATION,
product=product,
retry=Retry(maximum=10.0),
timeout=5,
task_id="product_create_2",
)
Templating¶
template_fields: Sequence[str] = (
"location",
"project_id",
"product_id",
"gcp_conn_id",
"impersonation_chain",
)
More information¶
CloudVisionDeleteProductOperator¶
Permanently deletes a product and its reference images.
Metadata of the product and all its images will be deleted right away, but search queries
against ProductSets
containing the product may still work until all related
caches are refreshed.
Possible errors:
Returns NOT_FOUND if the product does not exist.
For parameter definition, take a look at
CloudVisionDeleteProductOperator
Using the operator¶
If product_id
was generated by the API it can be extracted from XCOM:
product_delete = CloudVisionDeleteProductOperator(
location=LOCATION,
product_id="{{ task_instance.xcom_pull('product_create') }}",
task_id="product_delete",
)
Otherwise it can be specified explicitly:
product_delete_2 = CloudVisionDeleteProductOperator(
location=LOCATION, product_id=GCP_VISION_PRODUCT_ID, task_id="product_delete_2"
)
Templating¶
template_fields: Sequence[str] = (
"location",
"project_id",
"product_id",
"gcp_conn_id",
"impersonation_chain",
)
More information¶
CloudVisionGetProductOperator¶
Gets information associated with a Product
.
Possible errors:
Returns NOT_FOUND if the
Product
does not exist.
For parameter definition, take a look at
CloudVisionGetProductOperator
Using the operator¶
If product_id
was generated by the API it can be extracted from XCOM:
product_get = CloudVisionGetProductOperator(
location=LOCATION,
product_id="{{ task_instance.xcom_pull('product_create') }}",
task_id="product_get",
)
Otherwise it can be specified explicitly:
product_get_2 = CloudVisionGetProductOperator(
location=LOCATION, product_id=GCP_VISION_PRODUCT_ID, task_id="product_get_2"
)
Templating¶
template_fields: Sequence[str] = (
"location",
"project_id",
"product_id",
"gcp_conn_id",
"impersonation_chain",
)
More information¶
CloudVisionProductSetCreateOperator¶
Creates a new ProductSet
resource.
For parameter definition, take a look at
CloudVisionCreateProductSetOperator
Using the operator¶
We are using the ProductSet
and Retry
objects from Google libraries:
from google.cloud.vision_v1.types import ProductSet # isort:skip
from google.api_core.retry import Retry # isort:skip
product_set = ProductSet(display_name="My Product Set")
The product_set_id
argument can be omitted (it will be generated by the API):
product_set_create = CloudVisionCreateProductSetOperator(
location=LOCATION,
product_set=product_set,
retry=Retry(maximum=10.0),
timeout=5,
task_id="product_set_create",
)
Or it can be specified explicitly:
product_set_create_2 = CloudVisionCreateProductSetOperator(
product_set_id=GCP_VISION_PRODUCT_SET_ID,
location=LOCATION,
product_set=product_set,
retry=Retry(maximum=10.0),
timeout=5,
task_id="product_set_create_2",
)
Templating¶
template_fields: Sequence[str] = (
"location",
"project_id",
"product_set_id",
"gcp_conn_id",
"impersonation_chain",
)
More information¶
CloudVisionDeleteProductSetOperator¶
Permanently deletes a ProductSet
. Products
and ReferenceImages
in
the ProductSet
are not deleted. The actual image files are not deleted from
Google Cloud Storage.
For parameter definition, take a look at
CloudVisionDeleteProductSetOperator
Using the operator¶
If product_set_id
was generated by the API it can be extracted from XCOM:
product_set_delete = CloudVisionDeleteProductSetOperator(
location=LOCATION,
product_set_id=product_set_create_output,
task_id="product_set_delete",
)
Otherwise it can be specified explicitly:
product_set_delete_2 = CloudVisionDeleteProductSetOperator(
location=LOCATION, product_set_id=GCP_VISION_PRODUCT_SET_ID, task_id="product_set_delete_2"
)
Templating¶
template_fields: Sequence[str] = (
"location",
"project_id",
"product_set_id",
"gcp_conn_id",
"impersonation_chain",
)
More information¶
CloudVisionGetProductSetOperator¶
Gets information associated with a ProductSet
.
For parameter definition, take a look at
CloudVisionGetProductSetOperator
Using the operator¶
If product_set_id
was generated by the API it can be extracted from XCOM:
product_set_get = CloudVisionGetProductSetOperator(
location=LOCATION,
product_set_id=product_set_create_output,
task_id="product_set_get",
)
Otherwise it can be specified explicitly:
product_set_get_2 = CloudVisionGetProductSetOperator(
location=LOCATION, product_set_id=GCP_VISION_PRODUCT_SET_ID, task_id="product_set_get_2"
)
Templating¶
template_fields: Sequence[str] = (
"location",
"project_id",
"product_set_id",
"gcp_conn_id",
"impersonation_chain",
)
More information¶
CloudVisionUpdateProductSetOperator¶
Makes changes to a ProductSet
resource. Only display_name
can be updated
currently.
Note
To locate the ProductSet
resource, its name
in the form
projects/PROJECT_ID/locations/LOC_ID/productSets/PRODUCT_SET_ID
is necessary.
You can provide the name
directly as an attribute of the product_set
object.
However, you can leave it blank and provide location
and product_set_id
instead (and
optionally project_id
- if not present, the connection default will be used) and the
name
will be created by the operator itself.
This mechanism exists for your convenience, to allow leaving the project_id
empty and
having Airflow use the connection default project_id
.
For parameter definition, take a look at
CloudVisionUpdateProductSetOperator
Using the operator¶
We are using the ProductSet
object from the Google Cloud Vision library:
from google.cloud.vision_v1.types import ProductSet # isort:skip
product_set = ProductSet(display_name="My Product Set")
Initialization of the task:
If product_set_id
was generated by the API it can be extracted from XCOM:
product_set_update = CloudVisionUpdateProductSetOperator(
location=LOCATION,
product_set_id=product_set_create_output,
product_set=ProductSet(display_name="My Product Set 2"),
task_id="product_set_update",
)
Otherwise it can be specified explicitly:
product_set_update_2 = CloudVisionUpdateProductSetOperator(
location=LOCATION,
product_set_id=GCP_VISION_PRODUCT_SET_ID,
product_set=ProductSet(display_name="My Product Set 2"),
task_id="product_set_update_2",
)
Templating¶
template_fields: Sequence[str] = (
"location",
"project_id",
"product_set_id",
"gcp_conn_id",
"impersonation_chain",
)
More information¶
CloudVisionUpdateProductOperator¶
Makes changes to a Product
resource. Only the display_name
,
description
, and labels
fields can be updated right now.
If labels are updated, the change will not be reflected in queries until the next index
time.
Note
To locate the Product
resource, its name
in the form
projects/PROJECT_ID/locations/LOC_ID/products/PRODUCT_ID
is necessary.
You can provide the name
directly as an attribute of the product
object. However, you
can leave it blank and provide location
and product_id
instead (and optionally
project_id
- if not present, the connection default will be used) and the name
will
be created by the operator itself.
This mechanism exists for your convenience, to allow leaving the project_id
empty and
having Airflow use the connection default project_id
.
Possible errors:
Returns NOT_FOUND if the
Product
does not exist.Returns INVALID_ARGUMENT if
display_name
is present inupdate_mask
but is missing from the request or longer than 4096 characters.Returns INVALID_ARGUMENT if
description
is present inupdate_mask
but is longer than 4096 characters.Returns INVALID_ARGUMENT if
product_category
is present inupdate_mask
.
For parameter definition, take a look at
CloudVisionUpdateProductOperator
Using the operator¶
We are using the Product
object from the Google Cloud Vision library:
from google.cloud.vision_v1.types import Product # isort:skip
product = Product(display_name="My Product 1", product_category="toys")
If product_id
was generated by the API it can be extracted from XCOM:
product_update = CloudVisionUpdateProductOperator(
location=LOCATION,
product_id="{{ task_instance.xcom_pull('product_create') }}",
product=Product(display_name="My Product 2", description="My updated description"),
task_id="product_update",
)
Otherwise it can be specified explicitly:
product_update_2 = CloudVisionUpdateProductOperator(
location=LOCATION,
product_id=GCP_VISION_PRODUCT_ID,
product=Product(display_name="My Product 2", description="My updated description"),
task_id="product_update_2",
)
Templating¶
template_fields: Sequence[str] = (
"location",
"project_id",
"product_id",
"gcp_conn_id",
"impersonation_chain",
)
More information¶
CloudVisionCreateReferenceImageOperator¶
Creates a new ReferenceImage
resource.
For parameter definition, take a look at
CloudVisionCreateReferenceImageOperator
Using the operator¶
We are using the ReferenceImage
and Retry
objects from Google libraries:
from google.cloud.vision_v1.types import ReferenceImage # isort:skip
from google.api_core.retry import Retry # isort:skip
reference_image = ReferenceImage(uri=VISION_IMAGE_URL)
The product_set_id
argument can be omitted (it will be generated by the API):
reference_image_create = CloudVisionCreateReferenceImageOperator(
location=LOCATION,
reference_image=reference_image,
product_id="{{ task_instance.xcom_pull('product_create') }}",
reference_image_id=GCP_VISION_REFERENCE_IMAGE_ID,
retry=Retry(maximum=10.0),
timeout=5,
task_id="reference_image_create",
)
Or it can be specified explicitly:
reference_image_create_2 = CloudVisionCreateReferenceImageOperator(
location=LOCATION,
reference_image=reference_image,
product_id=GCP_VISION_PRODUCT_ID,
reference_image_id=GCP_VISION_REFERENCE_IMAGE_ID,
retry=Retry(maximum=10.0),
timeout=5,
task_id="reference_image_create_2",
)
Templating¶
template_fields: Sequence[str] = (
"location",
"reference_image",
"product_id",
"reference_image_id",
"project_id",
"gcp_conn_id",
"impersonation_chain",
)
More information¶
See Google Cloud Vision ReferenceImage delete documentation.
CloudVisionDeleteReferenceImageOperator¶
Deletes a ReferenceImage
resource.
For parameter definition, take a look at
CloudVisionDeleteReferenceImageOperator
Using the operator¶
We are using the ReferenceImage
and Retry
objects from Google libraries:
from google.cloud.vision_v1.types import ReferenceImage # isort:skip
from google.api_core.retry import Retry # isort:skip
reference_image = ReferenceImage(uri=VISION_IMAGE_URL)
The product_set_id
argument can be omitted (it will be generated by the API):
reference_image_delete = CloudVisionDeleteReferenceImageOperator(
location=LOCATION,
product_id="{{ task_instance.xcom_pull('product_create') }}",
reference_image_id=GCP_VISION_REFERENCE_IMAGE_ID,
retry=Retry(maximum=10.0),
timeout=5,
task_id="reference_image_delete",
)
Or it can be specified explicitly:
reference_image_delete_2 = CloudVisionDeleteReferenceImageOperator(
location=LOCATION,
reference_image_id=GCP_VISION_REFERENCE_IMAGE_ID,
product_id=GCP_VISION_PRODUCT_ID,
retry=Retry(maximum=10.0),
timeout=5,
task_id="reference_image_delete_2",
)
Templating¶
template_fields: Sequence[str] = (
"location",
"reference_image",
"product_id",
"reference_image_id",
"project_id",
"gcp_conn_id",
"impersonation_chain",
)
More information¶
See Google Cloud Vision ReferenceImage create documentation.
CloudVisionRemoveProductFromProductSetOperator¶
Creates a new ReferenceImage
resource.
For parameter definition, take a look at
CloudVisionRemoveProductFromProductSetOperator
Using the operator¶
We are using the Product
,
ProductSet
and Retry
objects from
Google libraries:
from google.api_core.retry import Retry # isort:skip
from google.cloud.vision_v1.types import ProductSet # isort:skip
from google.cloud.vision_v1.types import Product # isort:skip
If product_set_id
and product_id
was generated by the API it can be extracted from XCOM:
remove_product_from_product_set = CloudVisionRemoveProductFromProductSetOperator(
location=LOCATION,
product_set_id=product_set_create_output,
product_id="{{ task_instance.xcom_pull('product_create') }}",
retry=Retry(maximum=10.0),
timeout=5,
task_id="remove_product_from_product_set",
)
Otherwise it can be specified explicitly:
remove_product_from_product_set_2 = CloudVisionRemoveProductFromProductSetOperator(
location=LOCATION,
product_set_id=GCP_VISION_PRODUCT_SET_ID,
product_id=GCP_VISION_PRODUCT_ID,
retry=Retry(maximum=10.0),
timeout=5,
task_id="remove_product_from_product_set_2",
)
Templating¶
template_fields: Sequence[str] = (
"location",
"product_set_id",
"product_id",
"project_id",
"gcp_conn_id",
"impersonation_chain",
)
More information¶
See Google Cloud Vision Remove Product From Product Set documentation.
CloudVisionDetectTextOperator¶
Run text detection for an image.
For parameter definition, take a look at
CloudVisionDetectTextOperator
Using the operator¶
We are using the Retry
objects from
Google libraries:
from google.api_core.retry import Retry # isort:skip
detect_text = CloudVisionDetectTextOperator(
image=DETECT_IMAGE,
retry=Retry(maximum=10.0),
timeout=5,
task_id="detect_text",
language_hints="en",
web_detection_params={"include_geo_results": True},
)
The result can be extracted from XCOM:
detect_text_result = BashOperator(
bash_command="echo {{ task_instance.xcom_pull('detect_text')['textAnnotations'][0] }}",
task_id="detect_text_result",
)
Templating¶
template_fields: Sequence[str] = (
"image",
"max_results",
"timeout",
"gcp_conn_id",
"impersonation_chain",
)
More information¶
See Google Cloud Vision Text Detection documentation.
CloudVisionTextDetectOperator¶
Run document text detection for an image.
For parameter definition, take a look at
CloudVisionTextDetectOperator
Using the operator¶
We are using the Retry
objects from
Google libraries:
from google.api_core.retry import Retry # isort:skip
document_detect_text = CloudVisionTextDetectOperator(
image=DETECT_IMAGE, retry=Retry(maximum=10.0), timeout=5, task_id="document_detect_text"
)
The result can be extracted from XCOM:
document_detect_text_result = BashOperator(
bash_command="echo {{ task_instance.xcom_pull('document_detect_text')['textAnnotations'][0] }}",
task_id="document_detect_text_result",
)
Templating¶
template_fields: Sequence[str] = (
"image",
"max_results",
"timeout",
"gcp_conn_id",
"impersonation_chain",
) # Iterable[str]
More information¶
See Google Cloud Vision Document Text Detection documentation.
CloudVisionDetectImageLabelsOperator¶
Run image label detection for an image.
For parameter definition, take a look at
CloudVisionDetectImageLabelsOperator
Using the operator¶
We are using the Retry
objects from
Google libraries:
from google.api_core.retry import Retry # isort:skip
detect_labels = CloudVisionDetectImageLabelsOperator(
image=DETECT_IMAGE, retry=Retry(maximum=10.0), timeout=5, task_id="detect_labels"
)
The result can be extracted from XCOM:
detect_labels_result = BashOperator(
bash_command="echo {{ task_instance.xcom_pull('detect_labels')['labelAnnotations'][0] }}",
task_id="detect_labels_result",
)
Templating¶
template_fields: Sequence[str] = (
"image",
"max_results",
"timeout",
"gcp_conn_id",
"impersonation_chain",
)
More information¶
See Google Cloud Vision Label Detection documentation.
CloudVisionDetectImageSafeSearchOperator¶
Run image label detection for an image.
For parameter definition, take a look at
CloudVisionDetectImageSafeSearchOperator
Using the operator¶
We are using the Retry
objects from
Google libraries:
from google.api_core.retry import Retry # isort:skip
detect_safe_search = CloudVisionDetectImageSafeSearchOperator(
image=DETECT_IMAGE, retry=Retry(maximum=10.0), timeout=5, task_id="detect_safe_search"
)
The result can be extracted from XCOM:
detect_safe_search_result = BashOperator(
bash_command=f"echo {detect_safe_search.output}",
task_id="detect_safe_search_result",
)
Templating¶
template_fields: Sequence[str] = (
"image",
"max_results",
"timeout",
"gcp_conn_id",
"impersonation_chain",
)
More information¶
See Google Cloud Vision Safe Search Detection documentation.