Managed backups

[MaxxleLLC]

Thank you for opening a Pull Request! Before submitting your PR, there are a few things you can do to make sure it goes smoothly:

• Make sure to open an issue as a bug/issue before writing your code! That way we can discuss the change, evaluate designs, and agree on the general idea
• Ensure the tests and linter pass
• Code coverage does not decrease (if any source code was changed)
• Appropriate docs were updated (if necessary)

Fixes #<issue_number_goes_here> 🦕

[AVaksman]

Might need to import _pb_timestamp_to_datetime to update expire_time and/or create_time from backup_pb message.

[AVaksman]

nit.
maybe call it cluter_id

[AVaksman]

api.update_backup() returns a backup_pb message, is it reasonable here to update _expire_time from the returned backup_pb message?

updated_backup_pb = api.update_backup(backup_update, update_mask)
self._expire_time = _pb_timestamp_to_datetime(updated_backup_pb._expire_time)

[AVaksman]

Suggesting re-using backup.from_pb method here which also verifies that backup_pb.name is as expected and sets other values start_time, end_time, size_bytes, state as well.

return [Backup.from_bp(backup_pb, self) for backup_pb in backup_list_pb]

[AVaksman]

Also suggesting to add list_backups to the Instance class as well to list backups for given instance as well.
If yes, then this method should delegate to the Instance.list_backups.

[AVaksman]

Suggesting re-using backup.from_pb method here which also verifies that backup_pb.name is as expected and sets other values start_time, end_time, size_bytes, state as well.

return [Backup.from_bp(backup_pb, self) for backup_pb in backup_list_pb]

Scratch above

Since gapic.list_backups returns a PageIterator, let's return an iterator to the user as well, without cashing all the data.

introduce a new method:

def _item_to_backup(self, iterator, backup_pb):
    # Convert a backup protobuf to a Backup object.

    return Backup.from_pb(backup_pb, self)

use the above method to set the item_to_value property on the page_iter returned by gapic.list_backups to be used on next

  backup_page_iter_pb = self._instance._client.table_admin_client.list_backups(
      self._instance.name + "/clusters/" + cluster_id,
      filter_=filter_,
      order_by=order_by,
      page_size=page_size,
  )

  backup_page_iter_pb.item_to_value = self._item_to_backup

  return page_iter

WDYT?

[AVaksman]

is table_id needed here? Can use self.table_id?
Use case

new_table = instance.table('non-existant-table')
op = new_table.restore(backup)
...

[mf2199]

This must be different from self.table_id, but probably worth renaming to new_table_id to avoid confusion.

[AVaksman]

This method returns a long running operation and does not update self. There is no need to instantiate a Table object before restore(). How about making this method into a @classmethod. WDYT?

[AVaksman]

Another suggestion, instead of cluster_id, backup_id, backup_name use param source which could be either full backup_name (projects/<project>/instances/<instance>/clusters/<cluster>/backups/<backup>) or Backup instance

from google.cloud.bigtable.backup import _BACKUP_NAME_RE
backup_name = ""
if not source:
  raise ValueError("")
if isinstance(source, Backup):
  backup_name = Backup.name
elif isinstance(source, str):
  match_backup_name = _BACKUP_NAME_RE.match(source)
  if match_backup_name is None:
    raise ValueError("")
  backup_name = source
else: raise ValueError("source should be ...")