Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions hcloud/storage_boxes/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,28 +2,34 @@

from .client import (
BoundStorageBox,
BoundStorageBoxSnapshot,
StorageBoxesClient,
StorageBoxesPageResult,
StorageBoxSnapshotsPageResult,
)
from .domain import (
CreateStorageBoxResponse,
DeleteStorageBoxResponse,
StorageBox,
StorageBoxAccessSettings,
StorageBoxFoldersResponse,
StorageBoxSnapshot,
StorageBoxSnapshotPlan,
StorageBoxStats,
)

__all__ = [
"BoundStorageBox",
"BoundStorageBoxSnapshot",
"CreateStorageBoxResponse",
"DeleteStorageBoxResponse",
"StorageBox",
"StorageBoxAccessSettings",
"StorageBoxesClient",
"StorageBoxesPageResult",
"StorageBoxFoldersResponse",
"StorageBoxSnapshot",
"StorageBoxSnapshotPlan",
"StorageBoxSnapshotsPageResult",
"StorageBoxStats",
]
229 changes: 229 additions & 0 deletions hcloud/storage_boxes/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,12 +8,15 @@
from ..storage_box_types import BoundStorageBoxType, StorageBoxType
from .domain import (
CreateStorageBoxResponse,
CreateStorageBoxSnapshotResponse,
DeleteStorageBoxResponse,
DeleteStorageBoxSnapshotResponse,
StorageBox,
StorageBoxAccessSettings,
StorageBoxFoldersResponse,
StorageBoxSnapshot,
StorageBoxSnapshotPlan,
StorageBoxSnapshotStats,
StorageBoxStats,
)

Expand Down Expand Up @@ -105,11 +108,42 @@ def get_actions(
# TODO: implement bound methods


class BoundStorageBoxSnapshot(BoundModelBase, StorageBoxSnapshot):
_client: StorageBoxesClient

model = StorageBoxSnapshot

def __init__(
self,
client: StorageBoxesClient,
data: dict[str, Any],
complete: bool = True,
):
raw = data.get("storage_box")
if raw is not None:
data["storage_box"] = BoundStorageBox(
client, data={"id": raw}, complete=False
)

raw = data.get("stats")
if raw is not None:
data["stats"] = StorageBoxSnapshotStats.from_dict(raw)

super().__init__(client, data, complete)

# TODO: implement bound methods


class StorageBoxesPageResult(NamedTuple):
storage_boxes: list[BoundStorageBox]
meta: Meta


class StorageBoxSnapshotsPageResult(NamedTuple):
snapshots: list[BoundStorageBoxSnapshot]
meta: Meta


class StorageBoxesClient(ResourceClientBase):
"""
A client for the Storage Boxes API.
Expand Down Expand Up @@ -556,3 +590,198 @@ def enable_snapshot_plan(
json=data,
)
return BoundAction(self._parent.actions, response["action"])

# Snapshots
###########################################################################

def get_snapshot_by_id(
self,
storage_box: StorageBox | BoundStorageBox,
id: int,
) -> BoundStorageBoxSnapshot:
"""
Returns a single Snapshot from a Storage Box.

See https://docs.hetzner.cloud/reference/hetzner#storage-box-snapshots-get-a-snapshot

:param storage_box: Storage Box to get the Snapshot from.
:param id: ID of the Snapshot.
"""
response = self._client.request(
method="GET",
url=f"{self._base_url}/{storage_box.id}/snapshots/{id}",
)
return BoundStorageBoxSnapshot(self, response["snapshot"])

def get_snapshot_by_name(
self,
storage_box: StorageBox | BoundStorageBox,
name: str,
) -> BoundStorageBoxSnapshot:
"""
Returns a single Snapshot from a Storage Box.

See https://docs.hetzner.cloud/reference/hetzner#storage-box-snapshots-list-snapshots

:param storage_box: Storage Box to get the Snapshot from.
:param name: Name of the Snapshot.
"""
return self._get_first_by(self.get_snapshot_list, storage_box, name=name)

def get_snapshot_list(
self,
storage_box: StorageBox | BoundStorageBox,
*,
name: str | None = None,
is_automatic: bool | None = None,
label_selector: str | None = None,
sort: list[str] | None = None,
) -> StorageBoxSnapshotsPageResult:
"""
Returns all Snapshots for a Storage Box.

See https://docs.hetzner.cloud/reference/hetzner#storage-box-snapshots-list-snapshots

:param storage_box: Storage Box to get the Snapshots from.
:param name: Filter resources by their name. The response will only contain the resources matching exactly the specified name.
:param is_automatic: Filter wether the snapshot was made by a Snapshot Plan.
:param label_selector: Filter resources by labels. The response will only contain resources matching the label selector.
:param sort: Sort resources by field and direction.
"""
params: dict[str, Any] = {}
if name is not None:
params["name"] = name
if is_automatic is not None:
params["is_automatic"] = is_automatic
if label_selector is not None:
params["label_selector"] = label_selector
if sort is not None:
params["sort"] = sort

response = self._client.request(
method="GET",
url=f"{self._base_url}/{storage_box.id}/snapshots",
params=params,
)
return StorageBoxSnapshotsPageResult(
snapshots=[
BoundStorageBoxSnapshot(self, item) for item in response["snapshots"]
],
meta=Meta.parse_meta(response),
)

def get_snapshot_all(
self,
storage_box: StorageBox | BoundStorageBox,
*,
name: str | None = None,
is_automatic: bool | None = None,
label_selector: str | None = None,
sort: list[str] | None = None,
) -> list[BoundStorageBoxSnapshot]:
"""
Returns all Snapshots for a Storage Box.

See https://docs.hetzner.cloud/reference/hetzner#storage-box-snapshots-list-snapshots

:param storage_box: Storage Box to get the Snapshots from.
:param name: Filter resources by their name. The response will only contain the resources matching exactly the specified name.
:param is_automatic: Filter wether the snapshot was made by a Snapshot Plan.
:param label_selector: Filter resources by labels. The response will only contain resources matching the label selector.
:param sort: Sort resources by field and direction.
"""
# The endpoint does not have pagination, forward to the list method.
result, _ = self.get_snapshot_list(
storage_box,
name=name,
is_automatic=is_automatic,
label_selector=label_selector,
sort=sort,
)
return result

def create_snapshot(
self,
storage_box: StorageBox | BoundStorageBox,
*,
description: str | None = None,
labels: dict[str, str] | None = None,
) -> CreateStorageBoxSnapshotResponse:
"""
Creates a Snapshot of the Storage Box.

See https://docs.hetzner.cloud/reference/hetzner#storage-box-snapshots-create-a-snapshot

:param storage_box: Storage Box to create a Snapshot from.
:param description: Description of the Snapshot.
:param labels: User-defined labels (key/value pairs) for the Resource.
"""
data: dict[str, Any] = {}
if description is not None:
data["description"] = description
if labels is not None:
data["labels"] = labels

response = self._client.request(
method="POST",
url=f"{self._base_url}/{storage_box.id}/snapshots",
json=data,
)
return CreateStorageBoxSnapshotResponse(
snapshot=BoundStorageBoxSnapshot(self, response["snapshot"]),
action=BoundAction(self._parent.actions, response["action"]),
)

def update_snapshot(
self,
snapshot: StorageBoxSnapshot | BoundStorageBoxSnapshot,
*,
description: str | None = None,
labels: dict[str, str] | None = None,
) -> BoundStorageBoxSnapshot:
"""
Updates a Storage Box Snapshot.

See https://docs.hetzner.cloud/reference/hetzner#storage-box-snapshots-update-a-snapshot

:param snapshot: Storage Box Snapshot to update.
:param description: Description of the Snapshot.
:param labels: User-defined labels (key/value pairs) for the Resource.
"""
if snapshot.storage_box is None:
raise ValueError("snapshot storage_box property is none")

data: dict[str, Any] = {}
if description is not None:
data["description"] = description
if labels is not None:
data["labels"] = labels

response = self._client.request(
method="PUT",
url=f"{self._base_url}/{snapshot.storage_box.id}/snapshots/{snapshot.id}",
json=data,
)
return BoundStorageBoxSnapshot(self, response["snapshot"])

def delete_snapshot(
self,
snapshot: StorageBoxSnapshot | BoundStorageBoxSnapshot,
) -> DeleteStorageBoxSnapshotResponse:
"""
Deletes a Storage Box Snapshot.

See https://docs.hetzner.cloud/reference/hetzner#storage-box-snapshots-delete-a-snapshot

:param snapshot: Storage Box Snapshot to delete.
"""
if snapshot.storage_box is None:
raise ValueError("snapshot storage_box property is none")

response = self._client.request(
method="DELETE",
url=f"{self._base_url}/{snapshot.storage_box.id}/snapshots/{snapshot.id}",
)
return DeleteStorageBoxSnapshotResponse(
action=BoundAction(self._parent.actions, response["action"]),
)
76 changes: 74 additions & 2 deletions hcloud/storage_boxes/domain.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
from ..storage_box_types import BoundStorageBoxType, StorageBoxType

if TYPE_CHECKING:
from .client import BoundStorageBox
from .client import BoundStorageBox, BoundStorageBoxSnapshot

StorageBoxStatus = Literal[
"active",
Expand Down Expand Up @@ -252,17 +252,89 @@ class StorageBoxSnapshot(BaseDomain, DomainIdentityMixin):
Storage Box Snapshot Domain.
"""

# TODO: full domain
__api_properties__ = (
"id",
"name",
"description",
"is_automatic",
"labels",
"storage_box",
"created",
"stats",
)
__slots__ = __api_properties__

def __init__(
self,
id: int | None = None,
name: str | None = None,
description: str | None = None,
is_automatic: bool | None = None,
labels: dict[str, str] | None = None,
storage_box: BoundStorageBox | StorageBox | None = None,
created: str | None = None,
stats: StorageBoxSnapshotStats | None = None,
):
self.id = id
self.name = name
self.description = description
self.is_automatic = is_automatic
self.labels = labels
self.storage_box = storage_box
self.created = isoparse(created) if created else None
self.stats = stats


class StorageBoxSnapshotStats(BaseDomain):
"""
Storage Box Snapshot Stats Domain.
"""

__api_properties__ = (
"size",
"size_filesystem",
)
__slots__ = __api_properties__

def __init__(
self,
size: int,
size_filesystem: int,
):
self.size = size
self.size_filesystem = size_filesystem


class CreateStorageBoxSnapshotResponse(BaseDomain):
"""
Create Storage Box Snapshot Response Domain.
"""

__api_properties__ = (
"snapshot",
"action",
)
__slots__ = __api_properties__

def __init__(
self,
snapshot: BoundStorageBoxSnapshot,
action: BoundAction,
):
self.snapshot = snapshot
self.action = action


class DeleteStorageBoxSnapshotResponse(BaseDomain):
"""
Delete Storage Box Snapshot Response Domain.
"""

__api_properties__ = ("action",)
__slots__ = __api_properties__

def __init__(
self,
action: BoundAction,
):
self.action = action
Loading