ansible/hetzner.hcloud
1
0
Fork 0
mirror of https://github.com/ansible-collections/hetzner.hcloud.git synced 2026-02-04 08:01:49 +00:00
Code Activity

chore(deps): update dependency hcloud to v2.6.0 (#686)

Browse source 
This PR contains the following updates:

| Package | Change | Age | Confidence |
|---|---|---|---|
| [hcloud](https://redirect.github.com/hetznercloud/hcloud-python)
([changelog](https://redirect.github.com/hetznercloud/hcloud-python/blob/main/CHANGELOG.md))
| `2.5.4` -> `2.6.0` |
[![age](https://developer.mend.io/api/mc/badges/age/pypi/hcloud/2.6.0?slim=true)](https://docs.renovatebot.com/merge-confidence/)
|
[![confidence](https://developer.mend.io/api/mc/badges/confidence/pypi/hcloud/2.5.4/2.6.0?slim=true)](https://docs.renovatebot.com/merge-confidence/)
|

---

### Release Notes

<details>
<summary>hetznercloud/hcloud-python (hcloud)</summary>

###
[`v2.6.0`](https://redirect.github.com/hetznercloud/hcloud-python/blob/HEAD/CHANGELOG.md#v260)

[Compare
Source](https://redirect.github.com/hetznercloud/hcloud-python/compare/v2.5.4...v2.6.0)

##### Features

- add category property to server type
([#&#8203;549](https://redirect.github.com/hetznercloud/hcloud-python/issues/549))

##### Bug Fixes

- rename `ClientEntityBase` to `ResourceClientBase`
([#&#8203;532](https://redirect.github.com/hetznercloud/hcloud-python/issues/532))

</details>

---

### Configuration

📅 **Schedule**: Branch creation - At any time (no schedule defined),
Automerge - At any time (no schedule defined).

🚦 **Automerge**: Disabled by config. Please merge this manually once you
are satisfied.

♻ **Rebasing**: Whenever PR becomes conflicted, or you tick the
rebase/retry checkbox.

🔕 **Ignore**: Close this PR and you won't be reminded about this update
again.

---

- [ ] <!-- rebase-check -->If you want to rebase/retry this PR, check
this box

---

This PR was generated by [Mend Renovate](https://mend.io/renovate/).
View the [repository job
log](https://developer.mend.io/github/ansible-collections/hetzner.hcloud).

<!--renovate-debug:eyJjcmVhdGVkSW5WZXIiOiI0MS45MS4xIiwidXBkYXRlZEluVmVyIjoiNDEuOTEuMSIsInRhcmdldEJyYW5jaCI6Im1haW4iLCJsYWJlbHMiOltdfQ==-->

---------

Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com>
Co-authored-by: jo <ljonas@riseup.net>
This commit is contained in:
renovate[bot] 2025-09-08 12:44:47 +02:00 • committed by GitHub
parent f2983603af
commit 923057c7b6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
24 changed files with 451 additions and 411 deletions
Download patch file Download diff file Expand all files Collapse all files

4
plugins/module_utils/client.py
View file

@ -101,8 +101,8 @@ class Client(ClientBase):
Cached response will not expire, therefore the cached client must not be used
for long living scopes.
"""
self._requests_session = CachedSession()
self._client._session = CachedSession()
try:
yield
finally:
self._requests_session = requests.Session()
self._client._session = requests.Session()

131
plugins/module_utils/vendor/hcloud/_client.py vendored
View file

@ -81,6 +81,25 @@ def exponential_backoff_function(
return func
def _build_user_agent(
application_name: str | None,
application_version: str | None,
) -> str:
"""Build the user agent of the hcloud-python instance with the user application name (if specified)
:return: The user agent of this hcloud-python instance
"""
parts = []
for name, version in [
(application_name, application_version),
("hcloud-python", __version__),
]:
if name is not None:
parts.append(name if version is None else f"{name}/{version}")
return " ".join(parts)
class Client:
"""
Client for the Hetzner Cloud API.
@ -115,14 +134,6 @@ class Client:
breaking changes.
"""
_version = __version__
__user_agent_prefix = "hcloud-python"
_retry_interval = staticmethod(
exponential_backoff_function(base=1.0, multiplier=2, cap=60.0, jitter=True)
)
_retry_max_retries = 5
def __init__(
self,
token: str,
@ -146,18 +157,15 @@ class Client:
Max retries before timeout when polling actions from the API.
:param timeout: Requests timeout in seconds
"""
self.token = token
self._api_endpoint = api_endpoint
self._application_name = application_name
self._application_version = application_version
self._requests_session = requests.Session()
self._requests_timeout = timeout
if isinstance(poll_interval, (int, float)):
self._poll_interval_func = constant_backoff_function(poll_interval)
else:
self._poll_interval_func = poll_interval
self._poll_max_retries = poll_max_retries
self._client = ClientBase(
token=token,
endpoint=api_endpoint,
application_name=application_name,
application_version=application_version,
poll_interval=poll_interval,
poll_max_retries=poll_max_retries,
timeout=timeout,
)
self.datacenters = DatacentersClient(self)
"""DatacentersClient Instance
@ -249,27 +257,58 @@ class Client:
:type: :class:`PlacementGroupsClient <hcloud.placement_groups.client.PlacementGroupsClient>`
"""
def _get_user_agent(self) -> str:
"""Get the user agent of the hcloud-python instance with the user application name (if specified)
def request( # type: ignore[no-untyped-def]
self,
method: str,
url: str,
**kwargs,
) -> dict:
"""Perform a request to the Hetzner Cloud API.
:return: The user agent of this hcloud-python instance
:param method: Method to perform the request.
:param url: URL to perform the request.
:param timeout: Requests timeout in seconds.
"""
user_agents = []
for name, version in [
(self._application_name, self._application_version),
(self.__user_agent_prefix, self._version),
]:
if name is not None:
user_agents.append(name if version is None else f"{name}/{version}")
return self._client.request(method, url, **kwargs)
return " ".join(user_agents)
def _get_headers(self) -> dict:
headers = {
"User-Agent": self._get_user_agent(),
"Authorization": f"Bearer {self.token}",
class ClientBase:
def __init__(
self,
token: str,
*,
endpoint: str,
application_name: str | None = None,
application_version: str | None = None,
poll_interval: int | float | BackoffFunction = 1.0,
poll_max_retries: int = 120,
timeout: float | tuple[float, float] | None = None,
):
self._token = token
self._endpoint = endpoint
self._user_agent = _build_user_agent(application_name, application_version)
self._headers = {
"User-Agent": self._user_agent,
"Authorization": f"Bearer {self._token}",
"Accept": "application/json",
}
return headers
if isinstance(poll_interval, (int, float)):
poll_interval_func = constant_backoff_function(poll_interval)
else:
poll_interval_func = poll_interval
self._poll_interval_func = poll_interval_func
self._poll_max_retries = poll_max_retries
self._retry_interval_func = exponential_backoff_function(
base=1.0, multiplier=2, cap=60.0, jitter=True
)
self._retry_max_retries = 5
self._timeout = timeout
self._session = requests.Session()
def request( # type: ignore[no-untyped-def]
self,
@ -277,22 +316,22 @@ class Client:
url: str,
**kwargs,
) -> dict:
"""Perform a request to the Hetzner Cloud API, wrapper around requests.request
"""Perform a request to the provided URL.
:param method: HTTP Method to perform the Request
:param url: URL of the Endpoint
:param timeout: Requests timeout in seconds
:param method: Method to perform the request.
:param url: URL to perform the request.
:param timeout: Requests timeout in seconds.
:return: Response
"""
kwargs.setdefault("timeout", self._requests_timeout)
kwargs.setdefault("timeout", self._timeout)
url = self._api_endpoint + url
headers = self._get_headers()
url = self._endpoint + url
headers = self._headers
retries = 0
while True:
try:
response = self._requests_session.request(
response = self._session.request(
method=method,
url=url,
headers=headers,
@ -301,13 +340,13 @@ class Client:
return self._read_response(response)
except APIException as exception:
if retries < self._retry_max_retries and self._retry_policy(exception):
time.sleep(self._retry_interval(retries))
time.sleep(self._retry_interval_func(retries))
retries += 1
continue
raise
except requests.exceptions.Timeout:
if retries < self._retry_max_retries:
time.sleep(self._retry_interval(retries))
time.sleep(self._retry_interval_func(retries))
retries += 1
continue
raise

2
plugins/module_utils/vendor/hcloud/_version.py vendored
View file

@ -1,3 +1,3 @@
from __future__ import annotations
__version__ = "2.5.4" # x-releaser-pleaser-version
__version__ = "2.6.0" # x-releaser-pleaser-version

20
plugins/module_utils/vendor/hcloud/actions/client.py vendored
View file

@ -4,7 +4,7 @@ import time
import warnings
from typing import TYPE_CHECKING, Any, NamedTuple
from ..core import BoundModelBase, ClientEntityBase, Meta
from ..core import BoundModelBase, Meta, ResourceClientBase
from .domain import Action, ActionFailedException, ActionTimeoutException
if TYPE_CHECKING:
@ -50,11 +50,19 @@ class ActionsPageResult(NamedTuple):
meta: Meta
class ResourceActionsClient(ClientEntityBase):
class ResourceActionsClient(ResourceClientBase):
_resource: str
def __init__(self, client: Client, resource: str | None):
super().__init__(client)
def __init__(self, client: ResourceClientBase | Client, resource: str | None):
if isinstance(client, ResourceClientBase):
super().__init__(client._parent)
# Use the same base client as the the resource base client. Allows us to
# choose the base client outside of the ResourceActionsClient.
self._client = client._client
else:
# Backward compatibility, defaults to the parent ("top level") base client (`_client`).
super().__init__(client)
self._resource = resource or ""
def get_by_id(self, id: int) -> BoundAction:
@ -67,7 +75,7 @@ class ResourceActionsClient(ClientEntityBase):
url=f"{self._resource}/actions/{id}",
method="GET",
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def get_list(
self,
@ -104,7 +112,7 @@ class ResourceActionsClient(ClientEntityBase):
params=params,
)
actions = [
BoundAction(self._client.actions, action_data)
BoundAction(self._parent.actions, action_data)
for action_data in response["actions"]
]
return ActionsPageResult(actions, Meta.parse_meta(response))

32
plugins/module_utils/vendor/hcloud/certificates/client.py vendored
View file

@ -3,7 +3,7 @@ from __future__ import annotations
from typing import TYPE_CHECKING, Any, NamedTuple
from ..actions import ActionsPageResult, BoundAction, ResourceActionsClient
from ..core import BoundModelBase, ClientEntityBase, Meta
from ..core import BoundModelBase, Meta, ResourceClientBase
from .domain import (
Certificate,
CreateManagedCertificateResponse,
@ -103,8 +103,8 @@ class CertificatesPageResult(NamedTuple):
meta: Meta
class CertificatesClient(ClientEntityBase):
_client: Client
class CertificatesClient(ResourceClientBase):
_base_url = "/certificates"
actions: ResourceActionsClient
"""Certificates scoped actions client
@ -114,7 +114,7 @@ class CertificatesClient(ClientEntityBase):
def __init__(self, client: Client):
super().__init__(client)
self.actions = ResourceActionsClient(client, "/certificates")
self.actions = ResourceActionsClient(client, self._base_url)
def get_by_id(self, id: int) -> BoundCertificate:
"""Get a specific certificate by its ID.
@ -122,7 +122,7 @@ class CertificatesClient(ClientEntityBase):
:param id: int
:return: :class:`BoundCertificate <hcloud.certificates.client.BoundCertificate>`
"""
response = self._client.request(url=f"/certificates/{id}", method="GET")
response = self._client.request(url=f"{self._base_url}/{id}", method="GET")
return BoundCertificate(self, response["certificate"])
def get_list(
@ -157,9 +157,7 @@ class CertificatesClient(ClientEntityBase):
if per_page is not None:
params["per_page"] = per_page
response = self._client.request(
url="/certificates", method="GET", params=params
)
response = self._client.request(url=self._base_url, method="GET", params=params)
certificates = [
BoundCertificate(self, certificate_data)
@ -219,7 +217,7 @@ class CertificatesClient(ClientEntityBase):
}
if labels is not None:
data["labels"] = labels
response = self._client.request(url="/certificates", method="POST", json=data)
response = self._client.request(url=self._base_url, method="POST", json=data)
return BoundCertificate(self, response["certificate"])
def create_managed(
@ -245,10 +243,10 @@ class CertificatesClient(ClientEntityBase):
}
if labels is not None:
data["labels"] = labels
response = self._client.request(url="/certificates", method="POST", json=data)
response = self._client.request(url=self._base_url, method="POST", json=data)
return CreateManagedCertificateResponse(
certificate=BoundCertificate(self, response["certificate"]),
action=BoundAction(self._client.actions, response["action"]),
action=BoundAction(self._parent.actions, response["action"]),
)
def update(
@ -272,7 +270,7 @@ class CertificatesClient(ClientEntityBase):
if labels is not None:
data["labels"] = labels
response = self._client.request(
url=f"/certificates/{certificate.id}",
url=f"{self._base_url}/{certificate.id}",
method="PUT",
json=data,
)
@ -285,7 +283,7 @@ class CertificatesClient(ClientEntityBase):
:return: True
"""
self._client.request(
url=f"/certificates/{certificate.id}",
url=f"{self._base_url}/{certificate.id}",
method="DELETE",
)
# Return always true, because the API does not return an action for it. When an error occurs a HcloudAPIException will be raised
@ -323,12 +321,12 @@ class CertificatesClient(ClientEntityBase):
params["per_page"] = per_page
response = self._client.request(
url=f"/certificates/{certificate.id}/actions",
url=f"{self._base_url}/{certificate.id}/actions",
method="GET",
params=params,
)
actions = [
BoundAction(self._client.actions, action_data)
BoundAction(self._parent.actions, action_data)
for action_data in response["actions"]
]
return ActionsPageResult(actions, Meta.parse_meta(response))
@ -365,7 +363,7 @@ class CertificatesClient(ClientEntityBase):
:return: :class:`BoundAction <hcloud.actions.client.BoundAction>`
"""
response = self._client.request(
url=f"/certificates/{certificate.id}/actions/retry",
url=f"{self._base_url}/{certificate.id}/actions/retry",
method="POST",
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])

5
plugins/module_utils/vendor/hcloud/core/__init__.py vendored
View file

@ -1,13 +1,14 @@
from __future__ import annotations
from .client import BoundModelBase, ClientEntityBase
from .client import BoundModelBase, ClientEntityBase, ResourceClientBase
from .domain import BaseDomain, DomainIdentityMixin, Meta, Pagination
__all__ = [
"BaseDomain",
"BoundModelBase",
"ClientEntityBase",
"BaseDomain",
"DomainIdentityMixin",
"Meta",
"Pagination",
"ResourceClientBase",
]

42
plugins/module_utils/vendor/hcloud/core/client.py vendored
View file

@ -1,22 +1,24 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any, Callable
import warnings
from typing import TYPE_CHECKING, Any, Callable, ClassVar
if TYPE_CHECKING:
from .._client import Client
from .._client import Client, ClientBase
from .domain import BaseDomain
class ClientEntityBase:
_client: Client
class ResourceClientBase:
_base_url: ClassVar[str]
_parent: Client
_client: ClientBase
max_per_page: int = 50
def __init__(self, client: Client):
"""
:param client: Client
:return self
"""
self._client = client
self._parent = client
# Use the parent "default" base client.
self._client = client._client
def _iter_pages( # type: ignore[no-untyped-def]
self,
@ -50,14 +52,32 @@ class ClientEntityBase:
return entities[0] if entities else None
class ClientEntityBase(ResourceClientBase):
"""
Kept for backward compatibility.
.. deprecated:: 2.6.0
Use :class:``hcloud.core.client.ResourceClientBase`` instead.
"""
def __init__(self, client: Client):
warnings.warn(
"The 'hcloud.core.client.ClientEntityBase' class is deprecated, please use the "
"'hcloud.core.client.ResourceClientBase' class instead.",
DeprecationWarning,
stacklevel=2,
)
super().__init__(client)
class BoundModelBase:
"""Bound Model Base"""
model: Any
model: type[BaseDomain]
def __init__(
self,
client: ClientEntityBase,
client: ResourceClientBase,
data: dict,
complete: bool = True,
):

23
plugins/module_utils/vendor/hcloud/datacenters/client.py vendored
View file

@ -1,15 +1,12 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any, NamedTuple
from typing import Any, NamedTuple
from ..core import BoundModelBase, ClientEntityBase, Meta
from ..core import BoundModelBase, Meta, ResourceClientBase
from ..locations import BoundLocation
from ..server_types import BoundServerType
from .domain import Datacenter, DatacenterServerTypes
if TYPE_CHECKING:
from .._client import Client
class BoundDatacenter(BoundModelBase, Datacenter):
_client: DatacentersClient
@ -19,25 +16,25 @@ class BoundDatacenter(BoundModelBase, Datacenter):
def __init__(self, client: DatacentersClient, data: dict):
location = data.get("location")
if location is not None:
data["location"] = BoundLocation(client._client.locations, location)
data["location"] = BoundLocation(client._parent.locations, location)
server_types = data.get("server_types")
if server_types is not None:
available = [
BoundServerType(
client._client.server_types, {"id": server_type}, complete=False
client._parent.server_types, {"id": server_type}, complete=False
)
for server_type in server_types["available"]
]
supported = [
BoundServerType(
client._client.server_types, {"id": server_type}, complete=False
client._parent.server_types, {"id": server_type}, complete=False
)
for server_type in server_types["supported"]
]
available_for_migration = [
BoundServerType(
client._client.server_types, {"id": server_type}, complete=False
client._parent.server_types, {"id": server_type}, complete=False
)
for server_type in server_types["available_for_migration"]
]
@ -55,8 +52,8 @@ class DatacentersPageResult(NamedTuple):
meta: Meta
class DatacentersClient(ClientEntityBase):
_client: Client
class DatacentersClient(ResourceClientBase):
_base_url = "/datacenters"
def get_by_id(self, id: int) -> BoundDatacenter:
"""Get a specific datacenter by its ID.
@ -64,7 +61,7 @@ class DatacentersClient(ClientEntityBase):
:param id: int
:return: :class:`BoundDatacenter <hcloud.datacenters.client.BoundDatacenter>`
"""
response = self._client.request(url=f"/datacenters/{id}", method="GET")
response = self._client.request(url=f"{self._base_url}/{id}", method="GET")
return BoundDatacenter(self, response["datacenter"])
def get_list(
@ -93,7 +90,7 @@ class DatacentersClient(ClientEntityBase):
if per_page is not None:
params["per_page"] = per_page
response = self._client.request(url="/datacenters", method="GET", params=params)
response = self._client.request(url=self._base_url, method="GET", params=params)
datacenters = [
BoundDatacenter(self, datacenter_data)

40
plugins/module_utils/vendor/hcloud/firewalls/client.py vendored
View file

@ -3,7 +3,7 @@ from __future__ import annotations
from typing import TYPE_CHECKING, Any, NamedTuple
from ..actions import ActionsPageResult, BoundAction, ResourceActionsClient
from ..core import BoundModelBase, ClientEntityBase, Meta
from ..core import BoundModelBase, Meta, ResourceClientBase
from .domain import (
CreateFirewallResponse,
Firewall,
@ -52,7 +52,7 @@ class BoundFirewall(BoundModelBase, Firewall):
type=resource["type"],
server=(
BoundServer(
client._client.servers,
client._parent.servers,
resource.get("server"),
complete=False,
)
@ -68,7 +68,7 @@ class BoundFirewall(BoundModelBase, Firewall):
FirewallResource(
type=firewall_resource["type"],
server=BoundServer(
client._client.servers,
client._parent.servers,
firewall_resource["server"],
complete=False,
),
@ -183,8 +183,8 @@ class FirewallsPageResult(NamedTuple):
meta: Meta
class FirewallsClient(ClientEntityBase):
_client: Client
class FirewallsClient(ResourceClientBase):
_base_url = "/firewalls"
actions: ResourceActionsClient
"""Firewalls scoped actions client
@ -194,7 +194,7 @@ class FirewallsClient(ClientEntityBase):
def __init__(self, client: Client):
super().__init__(client)
self.actions = ResourceActionsClient(client, "/firewalls")
self.actions = ResourceActionsClient(client, self._base_url)
def get_actions_list(
self,
@ -227,12 +227,12 @@ class FirewallsClient(ClientEntityBase):
if per_page is not None:
params["per_page"] = per_page
response = self._client.request(
url=f"/firewalls/{firewall.id}/actions",
url=f"{self._base_url}/{firewall.id}/actions",
method="GET",
params=params,
)
actions = [
BoundAction(self._client.actions, action_data)
BoundAction(self._parent.actions, action_data)
for action_data in response["actions"]
]
return ActionsPageResult(actions, Meta.parse_meta(response))
@ -266,7 +266,7 @@ class FirewallsClient(ClientEntityBase):
:param id: int
:return: :class:`BoundFirewall <hcloud.firewalls.client.BoundFirewall>`
"""
response = self._client.request(url=f"/firewalls/{id}", method="GET")
response = self._client.request(url=f"{self._base_url}/{id}", method="GET")
return BoundFirewall(self, response["firewall"])
def get_list(
@ -303,7 +303,7 @@ class FirewallsClient(ClientEntityBase):
params["name"] = name
if sort is not None:
params["sort"] = sort
response = self._client.request(url="/firewalls", method="GET", params=params)
response = self._client.request(url=self._base_url, method="GET", params=params)
firewalls = [
BoundFirewall(self, firewall_data)
for firewall_data in response["firewalls"]
@ -373,12 +373,12 @@ class FirewallsClient(ClientEntityBase):
data.update({"apply_to": []})
for resource in resources:
data["apply_to"].append(resource.to_payload())
response = self._client.request(url="/firewalls", json=data, method="POST")
response = self._client.request(url=self._base_url, json=data, method="POST")
actions = []
if response.get("actions") is not None:
actions = [
BoundAction(self._client.actions, action_data)
BoundAction(self._parent.actions, action_data)
for action_data in response["actions"]
]
@ -409,7 +409,7 @@ class FirewallsClient(ClientEntityBase):
data["name"] = name
response = self._client.request(
url=f"/firewalls/{firewall.id}",
url=f"{self._base_url}/{firewall.id}",
method="PUT",
json=data,
)
@ -422,7 +422,7 @@ class FirewallsClient(ClientEntityBase):
:return: boolean
"""
self._client.request(
url=f"/firewalls/{firewall.id}",
url=f"{self._base_url}/{firewall.id}",
method="DELETE",
)
# Return always true, because the API does not return an action for it. When an error occurs a HcloudAPIException will be raised
@ -443,12 +443,12 @@ class FirewallsClient(ClientEntityBase):
for rule in rules:
data["rules"].append(rule.to_payload())
response = self._client.request(
url=f"/firewalls/{firewall.id}/actions/set_rules",
url=f"{self._base_url}/{firewall.id}/actions/set_rules",
method="POST",
json=data,
)
return [
BoundAction(self._client.actions, action_data)
BoundAction(self._parent.actions, action_data)
for action_data in response["actions"]
]
@ -467,12 +467,12 @@ class FirewallsClient(ClientEntityBase):
for resource in resources:
data["apply_to"].append(resource.to_payload())
response = self._client.request(
url=f"/firewalls/{firewall.id}/actions/apply_to_resources",
url=f"{self._base_url}/{firewall.id}/actions/apply_to_resources",
method="POST",
json=data,
)
return [
BoundAction(self._client.actions, action_data)
BoundAction(self._parent.actions, action_data)
for action_data in response["actions"]
]
@ -491,11 +491,11 @@ class FirewallsClient(ClientEntityBase):
for resource in resources:
data["remove_from"].append(resource.to_payload())
response = self._client.request(
url=f"/firewalls/{firewall.id}/actions/remove_from_resources",
url=f"{self._base_url}/{firewall.id}/actions/remove_from_resources",
method="POST",
json=data,
)
return [
BoundAction(self._client.actions, action_data)
BoundAction(self._parent.actions, action_data)
for action_data in response["actions"]
]

46
plugins/module_utils/vendor/hcloud/floating_ips/client.py vendored
View file

@ -3,7 +3,7 @@ from __future__ import annotations
from typing import TYPE_CHECKING, Any, NamedTuple
from ..actions import ActionsPageResult, BoundAction, ResourceActionsClient
from ..core import BoundModelBase, ClientEntityBase, Meta
from ..core import BoundModelBase, Meta, ResourceClientBase
from ..locations import BoundLocation
from .domain import CreateFloatingIPResponse, FloatingIP
@ -25,13 +25,13 @@ class BoundFloatingIP(BoundModelBase, FloatingIP):
server = data.get("server")
if server is not None:
data["server"] = BoundServer(
client._client.servers, {"id": server}, complete=False
client._parent.servers, {"id": server}, complete=False
)
home_location = data.get("home_location")
if home_location is not None:
data["home_location"] = BoundLocation(
client._client.locations, home_location
client._parent.locations, home_location
)
super().__init__(client, data, complete)
@ -139,8 +139,8 @@ class FloatingIPsPageResult(NamedTuple):
meta: Meta
class FloatingIPsClient(ClientEntityBase):
_client: Client
class FloatingIPsClient(ResourceClientBase):
_base_url = "/floating_ips"
actions: ResourceActionsClient
"""Floating IPs scoped actions client
@ -150,7 +150,7 @@ class FloatingIPsClient(ClientEntityBase):
def __init__(self, client: Client):
super().__init__(client)
self.actions = ResourceActionsClient(client, "/floating_ips")
self.actions = ResourceActionsClient(client, self._base_url)
def get_actions_list(
self,
@ -183,12 +183,12 @@ class FloatingIPsClient(ClientEntityBase):
if per_page is not None:
params["per_page"] = per_page
response = self._client.request(
url=f"/floating_ips/{floating_ip.id}/actions",
url=f"{self._base_url}/{floating_ip.id}/actions",
method="GET",
params=params,
)
actions = [
BoundAction(self._client.actions, action_data)
BoundAction(self._parent.actions, action_data)
for action_data in response["actions"]
]
return ActionsPageResult(actions, Meta.parse_meta(response))
@ -222,7 +222,7 @@ class FloatingIPsClient(ClientEntityBase):
:param id: int
:return: :class:`BoundFloatingIP <hcloud.floating_ips.client.BoundFloatingIP>`
"""
response = self._client.request(url=f"/floating_ips/{id}", method="GET")
response = self._client.request(url=f"{self._base_url}/{id}", method="GET")
return BoundFloatingIP(self, response["floating_ip"])
def get_list(
@ -255,9 +255,7 @@ class FloatingIPsClient(ClientEntityBase):
if name is not None:
params["name"] = name
response = self._client.request(
url="/floating_ips", method="GET", params=params
)
response = self._client.request(url=self._base_url, method="GET", params=params)
floating_ips = [
BoundFloatingIP(self, floating_ip_data)
for floating_ip_data in response["floating_ips"]
@ -325,11 +323,11 @@ class FloatingIPsClient(ClientEntityBase):
if name is not None:
data["name"] = name
response = self._client.request(url="/floating_ips", json=data, method="POST")
response = self._client.request(url=self._base_url, json=data, method="POST")
action = None
if response.get("action") is not None:
action = BoundAction(self._client.actions, response["action"])
action = BoundAction(self._parent.actions, response["action"])
result = CreateFloatingIPResponse(
floating_ip=BoundFloatingIP(self, response["floating_ip"]), action=action
@ -363,7 +361,7 @@ class FloatingIPsClient(ClientEntityBase):
data["name"] = name
response = self._client.request(
url=f"/floating_ips/{floating_ip.id}",
url=f"{self._base_url}/{floating_ip.id}",
method="PUT",
json=data,
)
@ -376,7 +374,7 @@ class FloatingIPsClient(ClientEntityBase):
:return: boolean
"""
self._client.request(
url=f"/floating_ips/{floating_ip.id}",
url=f"{self._base_url}/{floating_ip.id}",
method="DELETE",
)
# Return always true, because the API does not return an action for it. When an error occurs a HcloudAPIException will be raised
@ -399,11 +397,11 @@ class FloatingIPsClient(ClientEntityBase):
data.update({"delete": delete})
response = self._client.request(
url=f"/floating_ips/{floating_ip.id}/actions/change_protection",
url=f"{self._base_url}/{floating_ip.id}/actions/change_protection",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def assign(
self,
@ -418,11 +416,11 @@ class FloatingIPsClient(ClientEntityBase):
:return: :class:`BoundAction <hcloud.actions.client.BoundAction>`
"""
response = self._client.request(
url=f"/floating_ips/{floating_ip.id}/actions/assign",
url=f"{self._base_url}/{floating_ip.id}/actions/assign",
method="POST",
json={"server": server.id},
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def unassign(self, floating_ip: FloatingIP | BoundFloatingIP) -> BoundAction:
"""Unassigns a Floating IP, resulting in it being unreachable. You may assign it to a server again at a later time.
@ -431,10 +429,10 @@ class FloatingIPsClient(ClientEntityBase):
:return: :class:`BoundAction <hcloud.actions.client.BoundAction>`
"""
response = self._client.request(
url=f"/floating_ips/{floating_ip.id}/actions/unassign",
url=f"{self._base_url}/{floating_ip.id}/actions/unassign",
method="POST",
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def change_dns_ptr(
self,
@ -452,8 +450,8 @@ class FloatingIPsClient(ClientEntityBase):
:return: :class:`BoundAction <hcloud.actions.client.BoundAction>`
"""
response = self._client.request(
url=f"/floating_ips/{floating_ip.id}/actions/change_dns_ptr",
url=f"{self._base_url}/{floating_ip.id}/actions/change_dns_ptr",
method="POST",
json={"ip": ip, "dns_ptr": dns_ptr},
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])

28
plugins/module_utils/vendor/hcloud/images/client.py vendored
View file

@ -4,7 +4,7 @@ import warnings
from typing import TYPE_CHECKING, Any, NamedTuple
from ..actions import ActionsPageResult, BoundAction, ResourceActionsClient
from ..core import BoundModelBase, ClientEntityBase, Meta
from ..core import BoundModelBase, Meta, ResourceClientBase
from .domain import Image
if TYPE_CHECKING:
@ -23,12 +23,12 @@ class BoundImage(BoundModelBase, Image):
created_from = data.get("created_from")
if created_from is not None:
data["created_from"] = BoundServer(
client._client.servers, created_from, complete=False
client._parent.servers, created_from, complete=False
)
bound_to = data.get("bound_to")
if bound_to is not None:
data["bound_to"] = BoundServer(
client._client.servers, {"id": bound_to}, complete=False
client._parent.servers, {"id": bound_to}, complete=False
)
super().__init__(client, data)
@ -112,8 +112,8 @@ class ImagesPageResult(NamedTuple):
meta: Meta
class ImagesClient(ClientEntityBase):
_client: Client
class ImagesClient(ResourceClientBase):
_base_url = "/images"
actions: ResourceActionsClient
"""Images scoped actions client
@ -123,7 +123,7 @@ class ImagesClient(ClientEntityBase):
def __init__(self, client: Client):
super().__init__(client)
self.actions = ResourceActionsClient(client, "/images")
self.actions = ResourceActionsClient(client, self._base_url)
def get_actions_list(
self,
@ -156,12 +156,12 @@ class ImagesClient(ClientEntityBase):
if per_page is not None:
params["per_page"] = per_page
response = self._client.request(
url=f"/images/{image.id}/actions",
url=f"{self._base_url}/{image.id}/actions",
method="GET",
params=params,
)
actions = [
BoundAction(self._client.actions, action_data)
BoundAction(self._parent.actions, action_data)
for action_data in response["actions"]
]
return ActionsPageResult(actions, Meta.parse_meta(response))
@ -194,7 +194,7 @@ class ImagesClient(ClientEntityBase):
:param id: int
:return: :class:`BoundImage <hcloud.images.client.BoundImage`
"""
response = self._client.request(url=f"/images/{id}", method="GET")
response = self._client.request(url=f"{self._base_url}/{id}", method="GET")
return BoundImage(self, response["image"])
def get_list(
@ -255,7 +255,7 @@ class ImagesClient(ClientEntityBase):
params["status"] = per_page
if include_deprecated is not None:
params["include_deprecated"] = include_deprecated
response = self._client.request(url="/images", method="GET", params=params)
response = self._client.request(url=self._base_url, method="GET", params=params)
images = [BoundImage(self, image_data) for image_data in response["images"]]
return ImagesPageResult(images, Meta.parse_meta(response))
@ -371,7 +371,7 @@ class ImagesClient(ClientEntityBase):
if labels is not None:
data.update({"labels": labels})
response = self._client.request(
url=f"/images/{image.id}", method="PUT", json=data
url=f"{self._base_url}/{image.id}", method="PUT", json=data
)
return BoundImage(self, response["image"])
@ -381,7 +381,7 @@ class ImagesClient(ClientEntityBase):
:param :class:`BoundImage <hcloud.images.client.BoundImage>` or :class:`Image <hcloud.images.domain.Image>`
:return: bool
"""
self._client.request(url=f"/images/{image.id}", method="DELETE")
self._client.request(url=f"{self._base_url}/{image.id}", method="DELETE")
# Return allays true, because the API does not return an action for it. When an error occurs a APIException will be raised
return True
@ -402,8 +402,8 @@ class ImagesClient(ClientEntityBase):
data.update({"delete": delete})
response = self._client.request(
url=f"/images/{image.id}/actions/change_protection",
url=f"{self._base_url}/{image.id}/actions/change_protection",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])

15
plugins/module_utils/vendor/hcloud/isos/client.py vendored
View file

@ -1,13 +1,10 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any, NamedTuple
from typing import Any, NamedTuple
from ..core import BoundModelBase, ClientEntityBase, Meta
from ..core import BoundModelBase, Meta, ResourceClientBase
from .domain import Iso
if TYPE_CHECKING:
from .._client import Client
class BoundIso(BoundModelBase, Iso):
_client: IsosClient
@ -20,8 +17,8 @@ class IsosPageResult(NamedTuple):
meta: Meta
class IsosClient(ClientEntityBase):
_client: Client
class IsosClient(ResourceClientBase):
_base_url = "/isos"
def get_by_id(self, id: int) -> BoundIso:
"""Get a specific ISO by its id
@ -29,7 +26,7 @@ class IsosClient(ClientEntityBase):
:param id: int
:return: :class:`BoundIso <hcloud.isos.client.BoundIso>`
"""
response = self._client.request(url=f"/isos/{id}", method="GET")
response = self._client.request(url=f"{self._base_url}/{id}", method="GET")
return BoundIso(self, response["iso"])
def get_list(
@ -67,7 +64,7 @@ class IsosClient(ClientEntityBase):
if per_page is not None:
params["per_page"] = per_page
response = self._client.request(url="/isos", method="GET", params=params)
response = self._client.request(url=self._base_url, method="GET", params=params)
isos = [BoundIso(self, iso_data) for iso_data in response["isos"]]
return IsosPageResult(isos, Meta.parse_meta(response))

17
plugins/module_utils/vendor/hcloud/load_balancer_types/client.py vendored
View file

@ -1,13 +1,10 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any, NamedTuple
from typing import Any, NamedTuple
from ..core import BoundModelBase, ClientEntityBase, Meta
from ..core import BoundModelBase, Meta, ResourceClientBase
from .domain import LoadBalancerType
if TYPE_CHECKING:
from .._client import Client
class BoundLoadBalancerType(BoundModelBase, LoadBalancerType):
_client: LoadBalancerTypesClient
@ -20,8 +17,8 @@ class LoadBalancerTypesPageResult(NamedTuple):
meta: Meta
class LoadBalancerTypesClient(ClientEntityBase):
_client: Client
class LoadBalancerTypesClient(ResourceClientBase):
_base_url = "/load_balancer_types"
def get_by_id(self, id: int) -> BoundLoadBalancerType:
"""Returns a specific Load Balancer Type.
@ -30,7 +27,7 @@ class LoadBalancerTypesClient(ClientEntityBase):
:return: :class:`BoundLoadBalancerType <hcloud.load_balancer_type.client.BoundLoadBalancerType>`
"""
response = self._client.request(
url=f"/load_balancer_types/{id}",
url=f"{self._base_url}/{id}",
method="GET",
)
return BoundLoadBalancerType(self, response["load_balancer_type"])
@ -59,9 +56,7 @@ class LoadBalancerTypesClient(ClientEntityBase):
if per_page is not None:
params["per_page"] = per_page
response = self._client.request(
url="/load_balancer_types", method="GET", params=params
)
response = self._client.request(url=self._base_url, method="GET", params=params)
load_balancer_types = [
BoundLoadBalancerType(self, load_balancer_type_data)
for load_balancer_type_data in response["load_balancer_types"]

90
plugins/module_utils/vendor/hcloud/load_balancers/client.py vendored
View file

@ -10,7 +10,7 @@ except ImportError:
from ..actions import ActionsPageResult, BoundAction, ResourceActionsClient
from ..certificates import BoundCertificate
from ..core import BoundModelBase, ClientEntityBase, Meta
from ..core import BoundModelBase, Meta, ResourceClientBase
from ..load_balancer_types import BoundLoadBalancerType
from ..locations import BoundLocation
from ..metrics import Metrics
@ -67,7 +67,7 @@ class BoundLoadBalancer(BoundModelBase, LoadBalancer):
private_nets = [
PrivateNet(
network=BoundNetwork(
client._client.networks,
client._parent.networks,
{"id": private_net["network"]},
complete=False,
),
@ -84,7 +84,7 @@ class BoundLoadBalancer(BoundModelBase, LoadBalancer):
tmp_target = LoadBalancerTarget(type=target["type"])
if target["type"] == "server":
tmp_target.server = BoundServer(
client._client.servers, data=target["server"], complete=False
client._parent.servers, data=target["server"], complete=False
)
tmp_target.use_private_ip = target["use_private_ip"]
elif target["type"] == "label_selector":
@ -127,7 +127,7 @@ class BoundLoadBalancer(BoundModelBase, LoadBalancer):
)
tmp_service.http.certificates = [
BoundCertificate(
client._client.certificates,
client._parent.certificates,
{"id": certificate},
complete=False,
)
@ -155,12 +155,12 @@ class BoundLoadBalancer(BoundModelBase, LoadBalancer):
load_balancer_type = data.get("load_balancer_type")
if load_balancer_type is not None:
data["load_balancer_type"] = BoundLoadBalancerType(
client._client.load_balancer_types, load_balancer_type
client._parent.load_balancer_types, load_balancer_type
)
location = data.get("location")
if location is not None:
data["location"] = BoundLocation(client._client.locations, location)
data["location"] = BoundLocation(client._parent.locations, location)
super().__init__(client, data, complete)
@ -372,8 +372,8 @@ class LoadBalancersPageResult(NamedTuple):
meta: Meta
class LoadBalancersClient(ClientEntityBase):
_client: Client
class LoadBalancersClient(ResourceClientBase):
_base_url = "/load_balancers"
actions: ResourceActionsClient
"""Load Balancers scoped actions client
@ -383,7 +383,7 @@ class LoadBalancersClient(ClientEntityBase):
def __init__(self, client: Client):
super().__init__(client)
self.actions = ResourceActionsClient(client, "/load_balancers")
self.actions = ResourceActionsClient(client, self._base_url)
def get_by_id(self, id: int) -> BoundLoadBalancer:
"""Get a specific Load Balancer
@ -392,7 +392,7 @@ class LoadBalancersClient(ClientEntityBase):
:return: :class:`BoundLoadBalancer <hcloud.load_balancers.client.BoundLoadBalancer>`
"""
response = self._client.request(
url=f"/load_balancers/{id}",
url=f"{self._base_url}/{id}",
method="GET",
)
return BoundLoadBalancer(self, response["load_balancer"])
@ -426,9 +426,7 @@ class LoadBalancersClient(ClientEntityBase):
if per_page is not None:
params["per_page"] = per_page
response = self._client.request(
url="/load_balancers", method="GET", params=params
)
response = self._client.request(url=self._base_url, method="GET", params=params)
load_balancers = [
BoundLoadBalancer(self, load_balancer_data)
@ -518,11 +516,11 @@ class LoadBalancersClient(ClientEntityBase):
if location is not None:
data["location"] = location.id_or_name
response = self._client.request(url="/load_balancers", method="POST", json=data)
response = self._client.request(url=self._base_url, method="POST", json=data)
return CreateLoadBalancerResponse(
load_balancer=BoundLoadBalancer(self, response["load_balancer"]),
action=BoundAction(self._client.actions, response["action"]),
action=BoundAction(self._parent.actions, response["action"]),
)
def update(
@ -546,7 +544,7 @@ class LoadBalancersClient(ClientEntityBase):
if labels is not None:
data.update({"labels": labels})
response = self._client.request(
url=f"/load_balancers/{load_balancer.id}",
url=f"{self._base_url}/{load_balancer.id}",
method="PUT",
json=data,
)
@ -559,7 +557,7 @@ class LoadBalancersClient(ClientEntityBase):
:return: boolean
"""
self._client.request(
url=f"/load_balancers/{load_balancer.id}",
url=f"{self._base_url}/{load_balancer.id}",
method="DELETE",
)
return True
@ -596,7 +594,7 @@ class LoadBalancersClient(ClientEntityBase):
params["step"] = step
response = self._client.request(
url=f"/load_balancers/{load_balancer.id}/metrics",
url=f"{self._base_url}/{load_balancer.id}/metrics",
method="GET",
params=params,
)
@ -636,12 +634,12 @@ class LoadBalancersClient(ClientEntityBase):
params["per_page"] = per_page
response = self._client.request(
url=f"/load_balancers/{load_balancer.id}/actions",
url=f"{self._base_url}/{load_balancer.id}/actions",
method="GET",
params=params,
)
actions = [
BoundAction(self._client.actions, action_data)
BoundAction(self._parent.actions, action_data)
for action_data in response["actions"]
]
return ActionsPageResult(actions, Meta.parse_meta(response))
@ -683,11 +681,11 @@ class LoadBalancersClient(ClientEntityBase):
data: dict[str, Any] = service.to_payload()
response = self._client.request(
url=f"/load_balancers/{load_balancer.id}/actions/add_service",
url=f"{self._base_url}/{load_balancer.id}/actions/add_service",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def update_service(
self,
@ -703,11 +701,11 @@ class LoadBalancersClient(ClientEntityBase):
"""
data: dict[str, Any] = service.to_payload()
response = self._client.request(
url=f"/load_balancers/{load_balancer.id}/actions/update_service",
url=f"{self._base_url}/{load_balancer.id}/actions/update_service",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def delete_service(
self,
@ -724,11 +722,11 @@ class LoadBalancersClient(ClientEntityBase):
data: dict[str, Any] = {"listen_port": service.listen_port}
response = self._client.request(
url=f"/load_balancers/{load_balancer.id}/actions/delete_service",
url=f"{self._base_url}/{load_balancer.id}/actions/delete_service",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def add_target(
self,
@ -745,11 +743,11 @@ class LoadBalancersClient(ClientEntityBase):
data: dict[str, Any] = target.to_payload()
response = self._client.request(
url=f"/load_balancers/{load_balancer.id}/actions/add_target",
url=f"{self._base_url}/{load_balancer.id}/actions/add_target",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def remove_target(
self,
@ -768,11 +766,11 @@ class LoadBalancersClient(ClientEntityBase):
data.pop("use_private_ip", None)
response = self._client.request(
url=f"/load_balancers/{load_balancer.id}/actions/remove_target",
url=f"{self._base_url}/{load_balancer.id}/actions/remove_target",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def change_algorithm(
self,
@ -789,11 +787,11 @@ class LoadBalancersClient(ClientEntityBase):
data: dict[str, Any] = {"type": algorithm.type}
response = self._client.request(
url=f"/load_balancers/{load_balancer.id}/actions/change_algorithm",
url=f"{self._base_url}/{load_balancer.id}/actions/change_algorithm",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def change_dns_ptr(
self,
@ -811,11 +809,11 @@ class LoadBalancersClient(ClientEntityBase):
"""
response = self._client.request(
url=f"/load_balancers/{load_balancer.id}/actions/change_dns_ptr",
url=f"{self._base_url}/{load_balancer.id}/actions/change_dns_ptr",
method="POST",
json={"ip": ip, "dns_ptr": dns_ptr},
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def change_protection(
self,
@ -834,11 +832,11 @@ class LoadBalancersClient(ClientEntityBase):
data.update({"delete": delete})
response = self._client.request(
url=f"/load_balancers/{load_balancer.id}/actions/change_protection",
url=f"{self._base_url}/{load_balancer.id}/actions/change_protection",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def attach_to_network(
self,
@ -859,11 +857,11 @@ class LoadBalancersClient(ClientEntityBase):
data.update({"ip": ip})
response = self._client.request(
url=f"/load_balancers/{load_balancer.id}/actions/attach_to_network",
url=f"{self._base_url}/{load_balancer.id}/actions/attach_to_network",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def detach_from_network(
self,
@ -878,11 +876,11 @@ class LoadBalancersClient(ClientEntityBase):
"""
data: dict[str, Any] = {"network": network.id}
response = self._client.request(
url=f"/load_balancers/{load_balancer.id}/actions/detach_from_network",
url=f"{self._base_url}/{load_balancer.id}/actions/detach_from_network",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def enable_public_interface(
self,
@ -896,10 +894,10 @@ class LoadBalancersClient(ClientEntityBase):
"""
response = self._client.request(
url=f"/load_balancers/{load_balancer.id}/actions/enable_public_interface",
url=f"{self._base_url}/{load_balancer.id}/actions/enable_public_interface",
method="POST",
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def disable_public_interface(
self,
@ -913,10 +911,10 @@ class LoadBalancersClient(ClientEntityBase):
"""
response = self._client.request(
url=f"/load_balancers/{load_balancer.id}/actions/disable_public_interface",
url=f"{self._base_url}/{load_balancer.id}/actions/disable_public_interface",
method="POST",
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def change_type(
self,
@ -932,8 +930,8 @@ class LoadBalancersClient(ClientEntityBase):
"""
data: dict[str, Any] = {"load_balancer_type": load_balancer_type.id_or_name}
response = self._client.request(
url=f"/load_balancers/{load_balancer.id}/actions/change_type",
url=f"{self._base_url}/{load_balancer.id}/actions/change_type",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])

15
plugins/module_utils/vendor/hcloud/locations/client.py vendored
View file

@ -1,13 +1,10 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any, NamedTuple
from typing import Any, NamedTuple
from ..core import BoundModelBase, ClientEntityBase, Meta
from ..core import BoundModelBase, Meta, ResourceClientBase
from .domain import Location
if TYPE_CHECKING:
from .._client import Client
class BoundLocation(BoundModelBase, Location):
_client: LocationsClient
@ -20,8 +17,8 @@ class LocationsPageResult(NamedTuple):
meta: Meta
class LocationsClient(ClientEntityBase):
_client: Client
class LocationsClient(ResourceClientBase):
_base_url = "/locations"
def get_by_id(self, id: int) -> BoundLocation:
"""Get a specific location by its ID.
@ -29,7 +26,7 @@ class LocationsClient(ClientEntityBase):
:param id: int
:return: :class:`BoundLocation <hcloud.locations.client.BoundLocation>`
"""
response = self._client.request(url=f"/locations/{id}", method="GET")
response = self._client.request(url=f"{self._base_url}/{id}", method="GET")
return BoundLocation(self, response["location"])
def get_list(
@ -56,7 +53,7 @@ class LocationsClient(ClientEntityBase):
if per_page is not None:
params["per_page"] = per_page
response = self._client.request(url="/locations", method="GET", params=params)
response = self._client.request(url=self._base_url, method="GET", params=params)
locations = [
BoundLocation(self, location_data)
for location_data in response["locations"]

48
plugins/module_utils/vendor/hcloud/networks/client.py vendored
View file

@ -3,7 +3,7 @@ from __future__ import annotations
from typing import TYPE_CHECKING, Any, NamedTuple
from ..actions import ActionsPageResult, BoundAction, ResourceActionsClient
from ..core import BoundModelBase, ClientEntityBase, Meta
from ..core import BoundModelBase, Meta, ResourceClientBase
from .domain import Network, NetworkRoute, NetworkSubnet
if TYPE_CHECKING:
@ -32,7 +32,7 @@ class BoundNetwork(BoundModelBase, Network):
servers = data.get("servers", [])
if servers is not None:
servers = [
BoundServer(client._client.servers, {"id": server}, complete=False)
BoundServer(client._parent.servers, {"id": server}, complete=False)
for server in servers
]
data["servers"] = servers
@ -166,8 +166,8 @@ class NetworksPageResult(NamedTuple):
meta: Meta
class NetworksClient(ClientEntityBase):
_client: Client
class NetworksClient(ResourceClientBase):
_base_url = "/networks"
actions: ResourceActionsClient
"""Networks scoped actions client
@ -177,7 +177,7 @@ class NetworksClient(ClientEntityBase):
def __init__(self, client: Client):
super().__init__(client)
self.actions = ResourceActionsClient(client, "/networks")
self.actions = ResourceActionsClient(client, self._base_url)
def get_by_id(self, id: int) -> BoundNetwork:
"""Get a specific network
@ -185,7 +185,7 @@ class NetworksClient(ClientEntityBase):
:param id: int
:return: :class:`BoundNetwork <hcloud.networks.client.BoundNetwork>`
"""
response = self._client.request(url=f"/networks/{id}", method="GET")
response = self._client.request(url=f"{self._base_url}/{id}", method="GET")
return BoundNetwork(self, response["network"])
def get_list(
@ -217,7 +217,7 @@ class NetworksClient(ClientEntityBase):
if per_page is not None:
params["per_page"] = per_page
response = self._client.request(url="/networks", method="GET", params=params)
response = self._client.request(url=self._base_url, method="GET", params=params)
networks = [
BoundNetwork(self, network_data) for network_data in response["networks"]
@ -301,7 +301,7 @@ class NetworksClient(ClientEntityBase):
if labels is not None:
data["labels"] = labels
response = self._client.request(url="/networks", method="POST", json=data)
response = self._client.request(url=self._base_url, method="POST", json=data)
return BoundNetwork(self, response["network"])
@ -335,7 +335,7 @@ class NetworksClient(ClientEntityBase):
data.update({"labels": labels})
response = self._client.request(
url=f"/networks/{network.id}",
url=f"{self._base_url}/{network.id}",
method="PUT",
json=data,
)
@ -347,7 +347,7 @@ class NetworksClient(ClientEntityBase):
:param network: :class:`BoundNetwork <hcloud.networks.client.BoundNetwork>` or :class:`Network <hcloud.networks.domain.Network>`
:return: boolean
"""
self._client.request(url=f"/networks/{network.id}", method="DELETE")
self._client.request(url=f"{self._base_url}/{network.id}", method="DELETE")
return True
def get_actions_list(
@ -382,12 +382,12 @@ class NetworksClient(ClientEntityBase):
params["per_page"] = per_page
response = self._client.request(
url=f"/networks/{network.id}/actions",
url=f"{self._base_url}/{network.id}/actions",
method="GET",
params=params,
)
actions = [
BoundAction(self._client.actions, action_data)
BoundAction(self._parent.actions, action_data)
for action_data in response["actions"]
]
return ActionsPageResult(actions, Meta.parse_meta(response))
@ -436,11 +436,11 @@ class NetworksClient(ClientEntityBase):
data["vswitch_id"] = subnet.vswitch_id
response = self._client.request(
url=f"/networks/{network.id}/actions/add_subnet",
url=f"{self._base_url}/{network.id}/actions/add_subnet",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def delete_subnet(
self,
@ -457,11 +457,11 @@ class NetworksClient(ClientEntityBase):
data: dict[str, Any] = {"ip_range": subnet.ip_range}
response = self._client.request(
url=f"/networks/{network.id}/actions/delete_subnet",
url=f"{self._base_url}/{network.id}/actions/delete_subnet",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def add_route(
self,
@ -481,11 +481,11 @@ class NetworksClient(ClientEntityBase):
}
response = self._client.request(
url=f"/networks/{network.id}/actions/add_route",
url=f"{self._base_url}/{network.id}/actions/add_route",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def delete_route(
self,
@ -505,11 +505,11 @@ class NetworksClient(ClientEntityBase):
}
response = self._client.request(
url=f"/networks/{network.id}/actions/delete_route",
url=f"{self._base_url}/{network.id}/actions/delete_route",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def change_ip_range(
self,
@ -526,11 +526,11 @@ class NetworksClient(ClientEntityBase):
data: dict[str, Any] = {"ip_range": ip_range}
response = self._client.request(
url=f"/networks/{network.id}/actions/change_ip_range",
url=f"{self._base_url}/{network.id}/actions/change_ip_range",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def change_protection(
self,
@ -549,8 +549,8 @@ class NetworksClient(ClientEntityBase):
data.update({"delete": delete})
response = self._client.request(
url=f"/networks/{network.id}/actions/change_protection",
url=f"{self._base_url}/{network.id}/actions/change_protection",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])

27
plugins/module_utils/vendor/hcloud/placement_groups/client.py vendored
View file

@ -1,14 +1,11 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any, NamedTuple
from typing import Any, NamedTuple
from ..actions import BoundAction
from ..core import BoundModelBase, ClientEntityBase, Meta
from ..core import BoundModelBase, Meta, ResourceClientBase
from .domain import CreatePlacementGroupResponse, PlacementGroup
if TYPE_CHECKING:
from .._client import Client
class BoundPlacementGroup(BoundModelBase, PlacementGroup):
_client: PlacementGroupsClient
@ -43,8 +40,8 @@ class PlacementGroupsPageResult(NamedTuple):
meta: Meta
class PlacementGroupsClient(ClientEntityBase):
_client: Client
class PlacementGroupsClient(ResourceClientBase):
_base_url = "/placement_groups"
def get_by_id(self, id: int) -> BoundPlacementGroup:
"""Returns a specific Placement Group object
@ -53,7 +50,7 @@ class PlacementGroupsClient(ClientEntityBase):
:return: :class:`BoundPlacementGroup <hcloud.placement_groups.client.BoundPlacementGroup>`
"""
response = self._client.request(
url=f"/placement_groups/{id}",
url=f"{self._base_url}/{id}",
method="GET",
)
return BoundPlacementGroup(self, response["placement_group"])
@ -96,9 +93,7 @@ class PlacementGroupsClient(ClientEntityBase):
params["sort"] = sort
if type is not None:
params["type"] = type
response = self._client.request(
url="/placement_groups", method="GET", params=params
)
response = self._client.request(url=self._base_url, method="GET", params=params)
placement_groups = [
BoundPlacementGroup(self, placement_group_data)
for placement_group_data in response["placement_groups"]
@ -158,13 +153,11 @@ class PlacementGroupsClient(ClientEntityBase):
data: dict[str, Any] = {"name": name, "type": type}
if labels is not None:
data["labels"] = labels
response = self._client.request(
url="/placement_groups", json=data, method="POST"
)
response = self._client.request(url=self._base_url, json=data, method="POST")
action = None
if response.get("action") is not None:
action = BoundAction(self._client.actions, response["action"])
action = BoundAction(self._parent.actions, response["action"])
result = CreatePlacementGroupResponse(
placement_group=BoundPlacementGroup(self, response["placement_group"]),
@ -195,7 +188,7 @@ class PlacementGroupsClient(ClientEntityBase):
data["name"] = name
response = self._client.request(
url=f"/placement_groups/{placement_group.id}",
url=f"{self._base_url}/{placement_group.id}",
method="PUT",
json=data,
)
@ -208,7 +201,7 @@ class PlacementGroupsClient(ClientEntityBase):
:return: boolean
"""
self._client.request(
url=f"/placement_groups/{placement_group.id}",
url=f"{self._base_url}/{placement_group.id}",
method="DELETE",
)
return True

38
plugins/module_utils/vendor/hcloud/primary_ips/client.py vendored
View file

@ -3,7 +3,7 @@ from __future__ import annotations
from typing import TYPE_CHECKING, Any, NamedTuple
from ..actions import BoundAction, ResourceActionsClient
from ..core import BoundModelBase, ClientEntityBase, Meta
from ..core import BoundModelBase, Meta, ResourceClientBase
from .domain import CreatePrimaryIPResponse, PrimaryIP
if TYPE_CHECKING:
@ -22,7 +22,7 @@ class BoundPrimaryIP(BoundModelBase, PrimaryIP):
datacenter = data.get("datacenter", {})
if datacenter:
data["datacenter"] = BoundDatacenter(client._client.datacenters, datacenter)
data["datacenter"] = BoundDatacenter(client._parent.datacenters, datacenter)
super().__init__(client, data, complete)
@ -97,8 +97,8 @@ class PrimaryIPsPageResult(NamedTuple):
meta: Meta
class PrimaryIPsClient(ClientEntityBase):
_client: Client
class PrimaryIPsClient(ResourceClientBase):
_base_url = "/primary_ips"
actions: ResourceActionsClient
"""Primary IPs scoped actions client
@ -108,7 +108,7 @@ class PrimaryIPsClient(ClientEntityBase):
def __init__(self, client: Client):
super().__init__(client)
self.actions = ResourceActionsClient(client, "/primary_ips")
self.actions = ResourceActionsClient(client, self._base_url)
def get_by_id(self, id: int) -> BoundPrimaryIP:
"""Returns a specific Primary IP object.
@ -116,7 +116,7 @@ class PrimaryIPsClient(ClientEntityBase):
:param id: int
:return: :class:`BoundPrimaryIP <hcloud.primary_ips.client.BoundPrimaryIP>`
"""
response = self._client.request(url=f"/primary_ips/{id}", method="GET")
response = self._client.request(url=f"{self._base_url}/{id}", method="GET")
return BoundPrimaryIP(self, response["primary_ip"])
def get_list(
@ -154,7 +154,7 @@ class PrimaryIPsClient(ClientEntityBase):
if ip is not None:
params["ip"] = ip
response = self._client.request(url="/primary_ips", method="GET", params=params)
response = self._client.request(url=self._base_url, method="GET", params=params)
primary_ips = [
BoundPrimaryIP(self, primary_ip_data)
for primary_ip_data in response["primary_ips"]
@ -221,11 +221,11 @@ class PrimaryIPsClient(ClientEntityBase):
if labels is not None:
data["labels"] = labels
response = self._client.request(url="/primary_ips", json=data, method="POST")
response = self._client.request(url=self._base_url, json=data, method="POST")
action = None
if response.get("action") is not None:
action = BoundAction(self._client.actions, response["action"])
action = BoundAction(self._parent.actions, response["action"])
result = CreatePrimaryIPResponse(
primary_ip=BoundPrimaryIP(self, response["primary_ip"]), action=action
@ -259,7 +259,7 @@ class PrimaryIPsClient(ClientEntityBase):
data["name"] = name
response = self._client.request(
url=f"/primary_ips/{primary_ip.id}",
url=f"{self._base_url}/{primary_ip.id}",
method="PUT",
json=data,
)
@ -272,7 +272,7 @@ class PrimaryIPsClient(ClientEntityBase):
:return: boolean
"""
self._client.request(
url=f"/primary_ips/{primary_ip.id}",
url=f"{self._base_url}/{primary_ip.id}",
method="DELETE",
)
# Return always true, because the API does not return an action for it. When an error occurs a HcloudAPIException will be raised
@ -295,11 +295,11 @@ class PrimaryIPsClient(ClientEntityBase):
data.update({"delete": delete})
response = self._client.request(
url=f"/primary_ips/{primary_ip.id}/actions/change_protection",
url=f"{self._base_url}/{primary_ip.id}/actions/change_protection",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def assign(
self,
@ -317,11 +317,11 @@ class PrimaryIPsClient(ClientEntityBase):
:return: :class:`BoundAction <hcloud.actions.client.BoundAction>`
"""
response = self._client.request(
url=f"/primary_ips/{primary_ip.id}/actions/assign",
url=f"{self._base_url}/{primary_ip.id}/actions/assign",
method="POST",
json={"assignee_id": assignee_id, "assignee_type": assignee_type},
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def unassign(self, primary_ip: PrimaryIP | BoundPrimaryIP) -> BoundAction:
"""Unassigns a Primary IP, resulting in it being unreachable. You may assign it to a server again at a later time.
@ -330,10 +330,10 @@ class PrimaryIPsClient(ClientEntityBase):
:return: :class:`BoundAction <hcloud.actions.client.BoundAction>`
"""
response = self._client.request(
url=f"/primary_ips/{primary_ip.id}/actions/unassign",
url=f"{self._base_url}/{primary_ip.id}/actions/unassign",
method="POST",
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def change_dns_ptr(
self,
@ -351,8 +351,8 @@ class PrimaryIPsClient(ClientEntityBase):
:return: :class:`BoundAction <hcloud.actions.client.BoundAction>`
"""
response = self._client.request(
url=f"/primary_ips/{primary_ip.id}/actions/change_dns_ptr",
url=f"{self._base_url}/{primary_ip.id}/actions/change_dns_ptr",
method="POST",
json={"ip": ip, "dns_ptr": dns_ptr},
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])

17
plugins/module_utils/vendor/hcloud/server_types/client.py vendored
View file

@ -1,13 +1,10 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any, NamedTuple
from typing import Any, NamedTuple
from ..core import BoundModelBase, ClientEntityBase, Meta
from ..core import BoundModelBase, Meta, ResourceClientBase
from .domain import ServerType
if TYPE_CHECKING:
from .._client import Client
class BoundServerType(BoundModelBase, ServerType):
_client: ServerTypesClient
@ -20,8 +17,8 @@ class ServerTypesPageResult(NamedTuple):
meta: Meta
class ServerTypesClient(ClientEntityBase):
_client: Client
class ServerTypesClient(ResourceClientBase):
_base_url = "/server_types"
def get_by_id(self, id: int) -> BoundServerType:
"""Returns a specific Server Type.
@ -29,7 +26,7 @@ class ServerTypesClient(ClientEntityBase):
:param id: int
:return: :class:`BoundServerType <hcloud.server_types.client.BoundServerType>`
"""
response = self._client.request(url=f"/server_types/{id}", method="GET")
response = self._client.request(url=f"{self._base_url}/{id}", method="GET")
return BoundServerType(self, response["server_type"])
def get_list(
@ -56,9 +53,7 @@ class ServerTypesClient(ClientEntityBase):
if per_page is not None:
params["per_page"] = per_page
response = self._client.request(
url="/server_types", method="GET", params=params
)
response = self._client.request(url=self._base_url, method="GET", params=params)
server_types = [
BoundServerType(self, server_type_data)
for server_type_data in response["server_types"]

5
plugins/module_utils/vendor/hcloud/server_types/domain.py vendored
View file

@ -15,6 +15,8 @@ class ServerType(BaseDomain, DomainIdentityMixin):
Unique identifier of the server type
:param description: str
Description of the server type
:param category: str
Category of the Server Type.
:param cores: int
Number of cpu cores a server of this type will have
:param memory: int
@ -42,6 +44,7 @@ class ServerType(BaseDomain, DomainIdentityMixin):
"id",
"name",
"description",
"category",
"cores",
"memory",
"disk",
@ -66,6 +69,7 @@ class ServerType(BaseDomain, DomainIdentityMixin):
id: int | None = None,
name: str | None = None,
description: str | None = None,
category: str | None = None,
cores: int | None = None,
memory: int | None = None,
disk: int | None = None,
@ -80,6 +84,7 @@ class ServerType(BaseDomain, DomainIdentityMixin):
self.id = id
self.name = name
self.description = description
self.category = category
self.cores = cores
self.memory = memory
self.disk = disk

148
plugins/module_utils/vendor/hcloud/servers/client.py vendored
View file

@ -9,7 +9,7 @@ except ImportError:
isoparse = None
from ..actions import ActionsPageResult, BoundAction, ResourceActionsClient
from ..core import BoundModelBase, ClientEntityBase, Meta
from ..core import BoundModelBase, Meta, ResourceClientBase
from ..datacenters import BoundDatacenter
from ..firewalls import BoundFirewall
from ..floating_ips import BoundFloatingIP
@ -60,28 +60,28 @@ class BoundServer(BoundModelBase, Server):
def __init__(self, client: ServersClient, data: dict, complete: bool = True):
datacenter = data.get("datacenter")
if datacenter is not None:
data["datacenter"] = BoundDatacenter(client._client.datacenters, datacenter)
data["datacenter"] = BoundDatacenter(client._parent.datacenters, datacenter)
volumes = data.get("volumes", [])
if volumes:
volumes = [
BoundVolume(client._client.volumes, {"id": volume}, complete=False)
BoundVolume(client._parent.volumes, {"id": volume}, complete=False)
for volume in volumes
]
data["volumes"] = volumes
image = data.get("image", None)
if image is not None:
data["image"] = BoundImage(client._client.images, image)
data["image"] = BoundImage(client._parent.images, image)
iso = data.get("iso", None)
if iso is not None:
data["iso"] = BoundIso(client._client.isos, iso)
data["iso"] = BoundIso(client._parent.isos, iso)
server_type = data.get("server_type")
if server_type is not None:
data["server_type"] = BoundServerType(
client._client.server_types, server_type
client._parent.server_types, server_type
)
public_net = data.get("public_net")
@ -93,7 +93,7 @@ class BoundServer(BoundModelBase, Server):
)
ipv4_primary_ip = (
BoundPrimaryIP(
client._client.primary_ips,
client._parent.primary_ips,
{"id": public_net["ipv4"]["id"]},
complete=False,
)
@ -107,7 +107,7 @@ class BoundServer(BoundModelBase, Server):
)
ipv6_primary_ip = (
BoundPrimaryIP(
client._client.primary_ips,
client._parent.primary_ips,
{"id": public_net["ipv6"]["id"]},
complete=False,
)
@ -116,14 +116,14 @@ class BoundServer(BoundModelBase, Server):
)
floating_ips = [
BoundFloatingIP(
client._client.floating_ips, {"id": floating_ip}, complete=False
client._parent.floating_ips, {"id": floating_ip}, complete=False
)
for floating_ip in public_net["floating_ips"]
]
firewalls = [
PublicNetworkFirewall(
BoundFirewall(
client._client.firewalls, {"id": firewall["id"]}, complete=False
client._parent.firewalls, {"id": firewall["id"]}, complete=False
),
status=firewall["status"],
)
@ -146,7 +146,7 @@ class BoundServer(BoundModelBase, Server):
private_nets = [
PrivateNet(
network=BoundNetwork(
client._client.networks,
client._parent.networks,
{"id": private_net["network"]},
complete=False,
),
@ -161,7 +161,7 @@ class BoundServer(BoundModelBase, Server):
placement_group = data.get("placement_group")
if placement_group:
placement_group = BoundPlacementGroup(
client._client.placement_groups, placement_group
client._parent.placement_groups, placement_group
)
data["placement_group"] = placement_group
@ -485,8 +485,8 @@ class ServersPageResult(NamedTuple):
meta: Meta
class ServersClient(ClientEntityBase):
_client: Client
class ServersClient(ResourceClientBase):
_base_url = "/servers"
actions: ResourceActionsClient
"""Servers scoped actions client
@ -496,7 +496,7 @@ class ServersClient(ClientEntityBase):
def __init__(self, client: Client):
super().__init__(client)
self.actions = ResourceActionsClient(client, "/servers")
self.actions = ResourceActionsClient(client, self._base_url)
def get_by_id(self, id: int) -> BoundServer:
"""Get a specific server
@ -504,7 +504,7 @@ class ServersClient(ClientEntityBase):
:param id: int
:return: :class:`BoundServer <hcloud.servers.client.BoundServer>`
"""
response = self._client.request(url=f"/servers/{id}", method="GET")
response = self._client.request(url=f"{self._base_url}/{id}", method="GET")
return BoundServer(self, response["server"])
def get_list(
@ -541,7 +541,7 @@ class ServersClient(ClientEntityBase):
if per_page is not None:
params["per_page"] = per_page
response = self._client.request(url="/servers", method="GET", params=params)
response = self._client.request(url=self._base_url, method="GET", params=params)
ass_servers = [
BoundServer(self, server_data) for server_data in response["servers"]
@ -668,13 +668,13 @@ class ServersClient(ClientEntityBase):
data_public_net["ipv6"] = public_net.ipv6.id
data["public_net"] = data_public_net
response = self._client.request(url="/servers", method="POST", json=data)
response = self._client.request(url=self._base_url, method="POST", json=data)
result = CreateServerResponse(
server=BoundServer(self, response["server"]),
action=BoundAction(self._client.actions, response["action"]),
action=BoundAction(self._parent.actions, response["action"]),
next_actions=[
BoundAction(self._client.actions, action)
BoundAction(self._parent.actions, action)
for action in response["next_actions"]
],
root_password=response["root_password"],
@ -713,12 +713,12 @@ class ServersClient(ClientEntityBase):
params["per_page"] = per_page
response = self._client.request(
url=f"/servers/{server.id}/actions",
url=f"{self._base_url}/{server.id}/actions",
method="GET",
params=params,
)
actions = [
BoundAction(self._client.actions, action_data)
BoundAction(self._parent.actions, action_data)
for action_data in response["actions"]
]
return ActionsPageResult(actions, Meta.parse_meta(response))
@ -766,7 +766,7 @@ class ServersClient(ClientEntityBase):
if labels is not None:
data.update({"labels": labels})
response = self._client.request(
url=f"/servers/{server.id}",
url=f"{self._base_url}/{server.id}",
method="PUT",
json=data,
)
@ -804,7 +804,7 @@ class ServersClient(ClientEntityBase):
params["step"] = step
response = self._client.request(
url=f"/servers/{server.id}/metrics",
url=f"{self._base_url}/{server.id}/metrics",
method="GET",
params=params,
)
@ -818,8 +818,10 @@ class ServersClient(ClientEntityBase):
:param server: :class:`BoundServer <hcloud.servers.client.BoundServer>` or :class:`Server <hcloud.servers.domain.Server>`
:return: :class:`BoundAction <hcloud.actions.client.BoundAction>`
"""
response = self._client.request(url=f"/servers/{server.id}", method="DELETE")
return BoundAction(self._client.actions, response["action"])
response = self._client.request(
url=f"{self._base_url}/{server.id}", method="DELETE"
)
return BoundAction(self._parent.actions, response["action"])
def power_off(self, server: Server | BoundServer) -> BoundAction:
"""Cuts power to the server. This forcefully stops it without giving the server operating system time to gracefully stop
@ -828,10 +830,10 @@ class ServersClient(ClientEntityBase):
:return: :class:`BoundAction <hcloud.actions.client.BoundAction>`
"""
response = self._client.request(
url=f"/servers/{server.id}/actions/poweroff",
url=f"{self._base_url}/{server.id}/actions/poweroff",
method="POST",
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def power_on(self, server: Server | BoundServer) -> BoundAction:
"""Starts a server by turning its power on.
@ -840,10 +842,10 @@ class ServersClient(ClientEntityBase):
:return: :class:`BoundAction <hcloud.actions.client.BoundAction>`
"""
response = self._client.request(
url=f"/servers/{server.id}/actions/poweron",
url=f"{self._base_url}/{server.id}/actions/poweron",
method="POST",
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def reboot(self, server: Server | BoundServer) -> BoundAction:
"""Reboots a server gracefully by sending an ACPI request.
@ -852,10 +854,10 @@ class ServersClient(ClientEntityBase):
:return: :class:`BoundAction <hcloud.actions.client.BoundAction>`
"""
response = self._client.request(
url=f"/servers/{server.id}/actions/reboot",
url=f"{self._base_url}/{server.id}/actions/reboot",
method="POST",
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def reset(self, server: Server | BoundServer) -> BoundAction:
"""Cuts power to a server and starts it again.
@ -864,10 +866,10 @@ class ServersClient(ClientEntityBase):
:return: :class:`BoundAction <hcloud.actions.client.BoundAction>`
"""
response = self._client.request(
url=f"/servers/{server.id}/actions/reset",
url=f"{self._base_url}/{server.id}/actions/reset",
method="POST",
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def shutdown(self, server: Server | BoundServer) -> BoundAction:
"""Shuts down a server gracefully by sending an ACPI shutdown request.
@ -876,10 +878,10 @@ class ServersClient(ClientEntityBase):
:return: :class:`BoundAction <hcloud.actions.client.BoundAction>`
"""
response = self._client.request(
url=f"/servers/{server.id}/actions/shutdown",
url=f"{self._base_url}/{server.id}/actions/shutdown",
method="POST",
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def reset_password(self, server: Server | BoundServer) -> ResetPasswordResponse:
"""Resets the root password. Only works for Linux systems that are running the qemu guest agent.
@ -888,11 +890,11 @@ class ServersClient(ClientEntityBase):
:return: :class:`ResetPasswordResponse <hcloud.servers.domain.ResetPasswordResponse>`
"""
response = self._client.request(
url=f"/servers/{server.id}/actions/reset_password",
url=f"{self._base_url}/{server.id}/actions/reset_password",
method="POST",
)
return ResetPasswordResponse(
action=BoundAction(self._client.actions, response["action"]),
action=BoundAction(self._parent.actions, response["action"]),
root_password=response["root_password"],
)
@ -916,11 +918,11 @@ class ServersClient(ClientEntityBase):
"upgrade_disk": upgrade_disk,
}
response = self._client.request(
url=f"/servers/{server.id}/actions/change_type",
url=f"{self._base_url}/{server.id}/actions/change_type",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def enable_rescue(
self,
@ -943,12 +945,12 @@ class ServersClient(ClientEntityBase):
data.update({"ssh_keys": ssh_keys})
response = self._client.request(
url=f"/servers/{server.id}/actions/enable_rescue",
url=f"{self._base_url}/{server.id}/actions/enable_rescue",
method="POST",
json=data,
)
return EnableRescueResponse(
action=BoundAction(self._client.actions, response["action"]),
action=BoundAction(self._parent.actions, response["action"]),
root_password=response["root_password"],
)
@ -959,10 +961,10 @@ class ServersClient(ClientEntityBase):
:return: :class:`BoundAction <hcloud.actions.client.BoundAction>`
"""
response = self._client.request(
url=f"/servers/{server.id}/actions/disable_rescue",
url=f"{self._base_url}/{server.id}/actions/disable_rescue",
method="POST",
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def create_image(
self,
@ -994,13 +996,13 @@ class ServersClient(ClientEntityBase):
data.update({"labels": labels})
response = self._client.request(
url=f"/servers/{server.id}/actions/create_image",
url=f"{self._base_url}/{server.id}/actions/create_image",
method="POST",
json=data,
)
return CreateImageResponse(
action=BoundAction(self._client.actions, response["action"]),
image=BoundImage(self._client.images, response["image"]),
action=BoundAction(self._parent.actions, response["action"]),
image=BoundImage(self._parent.images, response["image"]),
)
def rebuild(
@ -1017,13 +1019,13 @@ class ServersClient(ClientEntityBase):
"""
data: dict[str, Any] = {"image": image.id_or_name}
response = self._client.request(
url=f"/servers/{server.id}/actions/rebuild",
url=f"{self._base_url}/{server.id}/actions/rebuild",
method="POST",
json=data,
)
return RebuildResponse(
action=BoundAction(self._client.actions, response["action"]),
action=BoundAction(self._parent.actions, response["action"]),
root_password=response.get("root_password"),
)
@ -1034,10 +1036,10 @@ class ServersClient(ClientEntityBase):
:return: :class:`BoundAction <hcloud.actions.client.BoundAction>`
"""
response = self._client.request(
url=f"/servers/{server.id}/actions/enable_backup",
url=f"{self._base_url}/{server.id}/actions/enable_backup",
method="POST",
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def disable_backup(self, server: Server | BoundServer) -> BoundAction:
"""Disables the automatic backup option and deletes all existing Backups for a Server.
@ -1046,10 +1048,10 @@ class ServersClient(ClientEntityBase):
:return: :class:`BoundAction <hcloud.actions.client.BoundAction>`
"""
response = self._client.request(
url=f"/servers/{server.id}/actions/disable_backup",
url=f"{self._base_url}/{server.id}/actions/disable_backup",
method="POST",
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def attach_iso(
self,
@ -1064,11 +1066,11 @@ class ServersClient(ClientEntityBase):
"""
data: dict[str, Any] = {"iso": iso.id_or_name}
response = self._client.request(
url=f"/servers/{server.id}/actions/attach_iso",
url=f"{self._base_url}/{server.id}/actions/attach_iso",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def detach_iso(self, server: Server | BoundServer) -> BoundAction:
"""Detaches an ISO from a server.
@ -1077,10 +1079,10 @@ class ServersClient(ClientEntityBase):
:return: :class:`BoundAction <hcloud.actions.client.BoundAction>`
"""
response = self._client.request(
url=f"/servers/{server.id}/actions/detach_iso",
url=f"{self._base_url}/{server.id}/actions/detach_iso",
method="POST",
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def change_dns_ptr(
self,
@ -1099,11 +1101,11 @@ class ServersClient(ClientEntityBase):
"""
data: dict[str, Any] = {"ip": ip, "dns_ptr": dns_ptr}
response = self._client.request(
url=f"/servers/{server.id}/actions/change_dns_ptr",
url=f"{self._base_url}/{server.id}/actions/change_dns_ptr",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def change_protection(
self,
@ -1127,11 +1129,11 @@ class ServersClient(ClientEntityBase):
data.update({"rebuild": rebuild})
response = self._client.request(
url=f"/servers/{server.id}/actions/change_protection",
url=f"{self._base_url}/{server.id}/actions/change_protection",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def request_console(self, server: Server | BoundServer) -> RequestConsoleResponse:
"""Requests credentials for remote access via vnc over websocket to keyboard, monitor, and mouse for a server.
@ -1140,11 +1142,11 @@ class ServersClient(ClientEntityBase):
:return: :class:`RequestConsoleResponse <hcloud.servers.domain.RequestConsoleResponse>`
"""
response = self._client.request(
url=f"/servers/{server.id}/actions/request_console",
url=f"{self._base_url}/{server.id}/actions/request_console",
method="POST",
)
return RequestConsoleResponse(
action=BoundAction(self._client.actions, response["action"]),
action=BoundAction(self._parent.actions, response["action"]),
wss_url=response["wss_url"],
password=response["password"],
)
@ -1172,11 +1174,11 @@ class ServersClient(ClientEntityBase):
if alias_ips is not None:
data.update({"alias_ips": alias_ips})
response = self._client.request(
url=f"/servers/{server.id}/actions/attach_to_network",
url=f"{self._base_url}/{server.id}/actions/attach_to_network",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def detach_from_network(
self,
@ -1191,11 +1193,11 @@ class ServersClient(ClientEntityBase):
"""
data: dict[str, Any] = {"network": network.id}
response = self._client.request(
url=f"/servers/{server.id}/actions/detach_from_network",
url=f"{self._base_url}/{server.id}/actions/detach_from_network",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def change_alias_ips(
self,
@ -1213,11 +1215,11 @@ class ServersClient(ClientEntityBase):
"""
data: dict[str, Any] = {"network": network.id, "alias_ips": alias_ips}
response = self._client.request(
url=f"/servers/{server.id}/actions/change_alias_ips",
url=f"{self._base_url}/{server.id}/actions/change_alias_ips",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def add_to_placement_group(
self,
@ -1232,11 +1234,11 @@ class ServersClient(ClientEntityBase):
"""
data: dict[str, Any] = {"placement_group": placement_group.id}
response = self._client.request(
url=f"/servers/{server.id}/actions/add_to_placement_group",
url=f"{self._base_url}/{server.id}/actions/add_to_placement_group",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])
def remove_from_placement_group(self, server: Server | BoundServer) -> BoundAction:
"""Removes a server from a placement group.
@ -1245,7 +1247,7 @@ class ServersClient(ClientEntityBase):
:return: :class:`BoundAction <hcloud.actions.client.BoundAction>`
"""
response = self._client.request(
url=f"/servers/{server.id}/actions/remove_from_placement_group",
url=f"{self._base_url}/{server.id}/actions/remove_from_placement_group",
method="POST",
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])

21
plugins/module_utils/vendor/hcloud/ssh_keys/client.py vendored
View file

@ -1,13 +1,10 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any, NamedTuple
from typing import Any, NamedTuple
from ..core import BoundModelBase, ClientEntityBase, Meta
from ..core import BoundModelBase, Meta, ResourceClientBase
from .domain import SSHKey
if TYPE_CHECKING:
from .._client import Client
class BoundSSHKey(BoundModelBase, SSHKey):
_client: SSHKeysClient
@ -41,8 +38,8 @@ class SSHKeysPageResult(NamedTuple):
meta: Meta
class SSHKeysClient(ClientEntityBase):
_client: Client
class SSHKeysClient(ResourceClientBase):
_base_url = "/ssh_keys"
def get_by_id(self, id: int) -> BoundSSHKey:
"""Get a specific SSH Key by its ID
@ -50,7 +47,7 @@ class SSHKeysClient(ClientEntityBase):
:param id: int
:return: :class:`BoundSSHKey <hcloud.ssh_keys.client.BoundSSHKey>`
"""
response = self._client.request(url=f"/ssh_keys/{id}", method="GET")
response = self._client.request(url=f"{self._base_url}/{id}", method="GET")
return BoundSSHKey(self, response["ssh_key"])
def get_list(
@ -87,7 +84,7 @@ class SSHKeysClient(ClientEntityBase):
if per_page is not None:
params["per_page"] = per_page
response = self._client.request(url="/ssh_keys", method="GET", params=params)
response = self._client.request(url=self._base_url, method="GET", params=params)
ssh_keys = [
BoundSSHKey(self, server_data) for server_data in response["ssh_keys"]
@ -153,7 +150,7 @@ class SSHKeysClient(ClientEntityBase):
data: dict[str, Any] = {"name": name, "public_key": public_key}
if labels is not None:
data["labels"] = labels
response = self._client.request(url="/ssh_keys", method="POST", json=data)
response = self._client.request(url=self._base_url, method="POST", json=data)
return BoundSSHKey(self, response["ssh_key"])
def update(
@ -177,7 +174,7 @@ class SSHKeysClient(ClientEntityBase):
if labels is not None:
data["labels"] = labels
response = self._client.request(
url=f"/ssh_keys/{ssh_key.id}",
url=f"{self._base_url}/{ssh_key.id}",
method="PUT",
json=data,
)
@ -189,6 +186,6 @@ class SSHKeysClient(ClientEntityBase):
:param ssh_key: :class:`BoundSSHKey <hcloud.ssh_keys.client.BoundSSHKey>` or :class:`SSHKey <hcloud.ssh_keys.domain.SSHKey>`
:return: True
"""
self._client.request(url=f"/ssh_keys/{ssh_key.id}", method="DELETE")
self._client.request(url=f"{self._base_url}/{ssh_key.id}", method="DELETE")
# Return always true, because the API does not return an action for it. When an error occurs a HcloudAPIException will be raised
return True

46
plugins/module_utils/vendor/hcloud/volumes/client.py vendored
View file

@ -3,7 +3,7 @@ from __future__ import annotations
from typing import TYPE_CHECKING, Any, NamedTuple
from ..actions import ActionsPageResult, BoundAction, ResourceActionsClient
from ..core import BoundModelBase, ClientEntityBase, Meta
from ..core import BoundModelBase, Meta, ResourceClientBase
from ..locations import BoundLocation
from .domain import CreateVolumeResponse, Volume
@ -21,7 +21,7 @@ class BoundVolume(BoundModelBase, Volume):
def __init__(self, client: VolumesClient, data: dict, complete: bool = True):
location = data.get("location")
if location is not None:
data["location"] = BoundLocation(client._client.locations, location)
data["location"] = BoundLocation(client._parent.locations, location)
# pylint: disable=import-outside-toplevel
from ..servers import BoundServer
@ -29,7 +29,7 @@ class BoundVolume(BoundModelBase, Volume):
server = data.get("server")
if server is not None:
data["server"] = BoundServer(
client._client.servers, {"id": server}, complete=False
client._parent.servers, {"id": server}, complete=False
)
super().__init__(client, data, complete)
@ -135,8 +135,8 @@ class VolumesPageResult(NamedTuple):
meta: Meta
class VolumesClient(ClientEntityBase):
_client: Client
class VolumesClient(ResourceClientBase):
_base_url = "/volumes"
actions: ResourceActionsClient
"""Volumes scoped actions client
@ -146,7 +146,7 @@ class VolumesClient(ClientEntityBase):
def __init__(self, client: Client):
super().__init__(client)
self.actions = ResourceActionsClient(client, "/volumes")
self.actions = ResourceActionsClient(client, self._base_url)
def get_by_id(self, id: int) -> BoundVolume:
"""Get a specific volume by its id
@ -154,7 +154,7 @@ class VolumesClient(ClientEntityBase):
:param id: int
:return: :class:`BoundVolume <hcloud.volumes.client.BoundVolume>`
"""
response = self._client.request(url=f"/volumes/{id}", method="GET")
response = self._client.request(url=f"{self._base_url}/{id}", method="GET")
return BoundVolume(self, response["volume"])
def get_list(
@ -191,7 +191,7 @@ class VolumesClient(ClientEntityBase):
if per_page is not None:
params["per_page"] = per_page
response = self._client.request(url="/volumes", method="GET", params=params)
response = self._client.request(url=self._base_url, method="GET", params=params)
volumes = [
BoundVolume(self, volume_data) for volume_data in response["volumes"]
]
@ -271,13 +271,13 @@ class VolumesClient(ClientEntityBase):
if format is not None:
data["format"] = format
response = self._client.request(url="/volumes", json=data, method="POST")
response = self._client.request(url=self._base_url, json=data, method="POST")
result = CreateVolumeResponse(
volume=BoundVolume(self, response["volume"]),
action=BoundAction(self._client.actions, response["action"]),
action=BoundAction(self._parent.actions, response["action"]),
next_actions=[
BoundAction(self._client.actions, action)
BoundAction(self._parent.actions, action)
for action in response["next_actions"]
],
)
@ -315,12 +315,12 @@ class VolumesClient(ClientEntityBase):
params["per_page"] = per_page
response = self._client.request(
url=f"/volumes/{volume.id}/actions",
url=f"{self._base_url}/{volume.id}/actions",
method="GET",
params=params,
)
actions = [
BoundAction(self._client.actions, action_data)
BoundAction(self._parent.actions, action_data)
for action_data in response["actions"]
]
return ActionsPageResult(actions, Meta.parse_meta(response))
@ -368,7 +368,7 @@ class VolumesClient(ClientEntityBase):
if labels is not None:
data.update({"labels": labels})
response = self._client.request(
url=f"/volumes/{volume.id}",
url=f"{self._base_url}/{volume.id}",
method="PUT",
json=data,
)
@ -380,7 +380,7 @@ class VolumesClient(ClientEntityBase):
:param volume: :class:`BoundVolume <hcloud.volumes.client.BoundVolume>` or :class:`Volume <hcloud.volumes.domain.Volume>`
:return: boolean
"""
self._client.request(url=f"/volumes/{volume.id}", method="DELETE")
self._client.request(url=f"{self._base_url}/{volume.id}", method="DELETE")
return True
def resize(self, volume: Volume | BoundVolume, size: int) -> BoundAction:
@ -392,11 +392,11 @@ class VolumesClient(ClientEntityBase):
:return: :class:`BoundAction <hcloud.actions.client.BoundAction>`
"""
data = self._client.request(
url=f"/volumes/{volume.id}/actions/resize",
url=f"{self._base_url}/{volume.id}/actions/resize",
json={"size": size},
method="POST",
)
return BoundAction(self._client.actions, data["action"])
return BoundAction(self._parent.actions, data["action"])
def attach(
self,
@ -416,11 +416,11 @@ class VolumesClient(ClientEntityBase):
data["automount"] = automount
data = self._client.request(
url=f"/volumes/{volume.id}/actions/attach",
url=f"{self._base_url}/{volume.id}/actions/attach",
json=data,
method="POST",
)
return BoundAction(self._client.actions, data["action"])
return BoundAction(self._parent.actions, data["action"])
def detach(self, volume: Volume | BoundVolume) -> BoundAction:
"""Detaches a volume from the server it’s attached to. You may attach it to a server again at a later time.
@ -429,10 +429,10 @@ class VolumesClient(ClientEntityBase):
:return: :class:`BoundAction <hcloud.actions.client.BoundAction>`
"""
data = self._client.request(
url=f"/volumes/{volume.id}/actions/detach",
url=f"{self._base_url}/{volume.id}/actions/detach",
method="POST",
)
return BoundAction(self._client.actions, data["action"])
return BoundAction(self._parent.actions, data["action"])
def change_protection(
self,
@ -451,8 +451,8 @@ class VolumesClient(ClientEntityBase):
data.update({"delete": delete})
response = self._client.request(
url=f"/volumes/{volume.id}/actions/change_protection",
url=f"{self._base_url}/{volume.id}/actions/change_protection",
method="POST",
json=data,
)
return BoundAction(self._client.actions, response["action"])
return BoundAction(self._parent.actions, response["action"])