def store()

in src/mozmlops/cloud_storage_api_client.py [0:0]

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

    def store(self, data: bytes, storage_path: str) -> str:
        """
        Arguments:
        data (bytes): The data to be stored in the cloud.
        storage_path (str): The filepath where the data will be stored.

        Places a blob of data, represented in bytes,
        at a specific filepath within the GCS project and bucket specified
        when the CloudStorageAPIClient was initialized.
        """
        from google.cloud import storage
        from google.cloud.exceptions import GoogleCloudError

        client = storage.Client(project=self.gcs_project_name)

        # Raises an exception if the bucket name cannot be found
        bucket = client.get_bucket(self.gcs_bucket_name)

        blob = bucket.blob(storage_path)

        with io.BytesIO(data) as f:
            # TODO: Catch exceptions and report back.

            # Google recommends setting `if_generation_match=0` if the
            # object is expected to be new. We don't expect collisions,
            # so setting this to 0 seems good.
            try:
                blob.upload_from_file(f, if_generation_match=0)
                log_line = f"The model is stored at {storage_path}"
                logging.info(log_line)
            except GoogleCloudError as e:
                if e.code == 412:
                    raise Exception(
                        "The object you tried to upload is already in the GCS bucket. Currently, the .store() function's implementation dictates this behavior."
                    ).with_traceback(e.__traceback__)
                raise e

        return storage_path