def annotate_gcs()

in src/gcf/main.py [0:0]

• 40 lines of code
• 6 McCabe index (conditional complexity)

def annotate_gcs(cloud_event):
    """Annotate image dropped into GCS bucket.

    Terraform configuration subscribes this function for the events generated by GCS bucket.
    event_trigger {
        trigger_region = var.gcf_location # The trigger must be in the same location as the bucket
        event_type = "google.cloud.storage.object.v1.finalized"
        retry_policy = "RETRY_POLICY_RETRY"
        service_account_email = google_service_account.gcf-sa.email
        event_filters {
        attribute = "bucket"
        value = var.input-bucket
        }
    }

    Args:
        cloud_event: is event generated by GCS bucket when new object is created
    """

    # read notification event data, showing also variables which are not used in this code
    data = cloud_event.data
    event_id = cloud_event["id"]
    event_type = cloud_event["type"]
    src_bucket = data["bucket"]
    image_file_name = data["name"]
    logging.info(
        f"Received event {event_type} id={event_id} from {src_bucket} for file {image_file_name}"
    )
    # build a list of requested annotation features form environment variable
    features_list = []
    features_env = os.environ.get(FEATURES_ENV, None)
    if features_env is None:
        logging.warn(
            "Annotation features aren't defined in the environment variable %s",
            FEATURES_ENV,
        )
    else:
        logging.info(f"{event_id}: Env. features:{features_env}")
        features_list = build_features_list(features_env)
    logging.info(f"{event_id}: Annotating for features: {features_list}")
    # form environment variable retreive bucket name for results
    annotations_bucket = os.environ.get(ANNOTATIONS_BUCKET_ENV)
    if annotations_bucket is None:
        logging.error("%s is not defined.", ANNOTATIONS_BUCKET_ENV)
        return
    # create result file name:
    annotations_file_name = json_filename_for_image(image_file_name)
    vision_image = read_vision_image_from_gcs(src_bucket, image_file_name)
    if vision_image:
        logging.info(
            f"{event_id}: Loaded {image_file_name} as vision.Image, executing annotations."
        )
        json_result = annotate_image(vision_image, features_list)
        logging.info(f"{event_id}: Annotated image {image_file_name}")
        if json_result:
            logging.info(
                f"{event_id}: Saving JSON: {annotations_bucket}/{annotations_file_name}"
            )
            gcs_write(annotations_bucket, annotations_file_name, json_result)
    else:
        logging.error(f"{event_id}: Image {image_file_name} could not be read.")
    logging.info(f"Event {event_id} is processed")