def upload_to_gcs()

in infra-as-code/modules/ingest-pipeline/cf-feedback-generator/record.py [0:0]

• 16 lines of code
• 3 McCabe index (conditional complexity)

  def upload_to_gcs(self, bucket_name, filename):
    """Uploads a resource to GCS

    Args:
        bucket_name (string): Bucket ID where the filename needs to be uploaded
        filename (string): The name of the local file to upload
    """
    retry_policy = retry.Retry(
      initial = 30,
      maximum = 810.0,
      multiplier = 2
    )

    try:
      bucket = self.storage_client.bucket(bucket_name)
      blob_name = f'{str(datetime.now().year)}/{filename}'
      blob = bucket.blob(blob_name)

      path_to_file = '/tmp/'+filename
      if os.path.isfile(path_to_file): 
          blob.upload_from_filename(path_to_file, retry=retry_policy)
      print('Uploaded file into gcs')
    except Exception as e:
      raise e