def create_file_checkpoint()

in jupyter-gcs-contents-manager/gcs_contents_manager.py [0:0]

• 12 lines of code
• 2 McCabe index (conditional complexity)

  def create_file_checkpoint(self, content, format, path):
    checkpoint_id = 'checkpoint'
    blob = self.checkpoint_blob(checkpoint_id, path, create_if_missing=True)
    content_type = 'text/plain' if format == 'text' else 'application/octet-stream'
    # GCS doesn't allow specifying the key version, so drop it if present
    if blob.kms_key_name:
      blob._properties['kmsKeyName'] = re.split('/cryptoKeyVersions/\d+$',
                                                blob.kms_key_name)[0]
    blob.upload_from_string(content, content_type=content_type)
    return {
        'id': checkpoint_id,
        'last_modified': blob.updated,
    }