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
12 changes: 6 additions & 6 deletions hcloud/_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
import time
from http import HTTPStatus
from random import uniform
from typing import Protocol
from typing import Any, Protocol

import requests

Expand Down Expand Up @@ -72,7 +72,7 @@ def exponential_backoff_function(
"""

def func(retries: int) -> float:
interval = base * multiplier**retries # Exponential backoff
interval: float = base * multiplier**retries # Exponential backoff
interval = min(cap, interval) # Cap backoff
if jitter:
interval = uniform(base, interval) # Add jitter
Expand Down Expand Up @@ -292,7 +292,7 @@ def request( # type: ignore[no-untyped-def]
method: str,
url: str,
**kwargs,
) -> dict:
) -> dict[str, Any]:
"""Perform a request to the Hetzner Cloud API.

:param method: Method to perform the request.
Expand Down Expand Up @@ -345,7 +345,7 @@ def request( # type: ignore[no-untyped-def]
method: str,
url: str,
**kwargs,
) -> dict:
) -> dict[str, Any]:
"""Perform a request to the provided URL.

:param method: Method to perform the request.
Expand Down Expand Up @@ -381,7 +381,7 @@ def request( # type: ignore[no-untyped-def]
continue
raise

def _read_response(self, response: requests.Response) -> dict:
def _read_response(self, response: requests.Response) -> dict[str, Any]:
correlation_id = response.headers.get("X-Correlation-Id")
payload = {}
try:
Expand All @@ -404,7 +404,7 @@ def _read_response(self, response: requests.Response) -> dict:
correlation_id=correlation_id,
)

error: dict = payload["error"]
error: dict[str, Any] = payload["error"]
raise APIException(
code=error["code"],
message=error["message"],
Expand Down
2 changes: 1 addition & 1 deletion hcloud/actions/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
from .._client import Client


class BoundAction(BoundModelBase, Action):
class BoundAction(BoundModelBase[Action], Action):
_client: ActionsClient

model = Action
Expand Down
20 changes: 16 additions & 4 deletions hcloud/actions/domain.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
from __future__ import annotations

from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Any, TypedDict

from .._exceptions import HCloudException
from ..core import BaseDomain
Expand Down Expand Up @@ -49,8 +49,8 @@ def __init__(
progress: int | None = None,
started: str | None = None,
finished: str | None = None,
resources: list[dict] | None = None,
error: dict | None = None,
resources: list[ActionResource] | None = None,
error: ActionError | None = None,
):
self.id = id
self.command = command
Expand All @@ -63,6 +63,17 @@ def __init__(
self.error = error


class ActionResource(TypedDict):
id: int
type: str


class ActionError(TypedDict):
code: str
message: str
details: dict[str, Any]


class ActionException(HCloudException):
"""A generic action exception"""

Expand All @@ -80,7 +91,8 @@ def __init__(self, action: Action | BoundAction):

extras.append(action.error["code"])
else:
extras.append(action.command)
if action.command is not None:
extras.append(action.command)

extras.append(str(action.id))
message += f" ({', '.join(extras)})"
Expand Down
9 changes: 7 additions & 2 deletions hcloud/certificates/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,12 +15,17 @@
from .._client import Client


class BoundCertificate(BoundModelBase, Certificate):
class BoundCertificate(BoundModelBase[Certificate], Certificate):
_client: CertificatesClient

model = Certificate

def __init__(self, client: CertificatesClient, data: dict, complete: bool = True):
def __init__(
self,
client: CertificatesClient,
data: dict[str, Any],
complete: bool = True,
):
status = data.get("status")
if status is not None:
error_data = status.get("error")
Expand Down
36 changes: 25 additions & 11 deletions hcloud/core/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,16 @@

import warnings
from collections.abc import Callable
from typing import TYPE_CHECKING, Any, ClassVar
from typing import TYPE_CHECKING, Any, ClassVar, Generic, TypeVar

from .domain import BaseDomain

if TYPE_CHECKING:
from .._client import Client, ClientBase
from .domain import BaseDomain
from .domain import Meta


T = TypeVar("T")


class ResourceClientBase:
Expand All @@ -23,10 +28,10 @@ def __init__(self, client: Client):

def _iter_pages( # type: ignore[no-untyped-def]
self,
list_function: Callable,
list_function: Callable[..., tuple[list[T], Meta]],
*args,
**kwargs,
) -> list:
) -> list[T]:
results = []

page = 1
Expand All @@ -46,7 +51,12 @@ def _iter_pages( # type: ignore[no-untyped-def]

return results

def _get_first_by(self, list_function: Callable, *args, **kwargs): # type: ignore[no-untyped-def]
def _get_first_by( # type: ignore[no-untyped-def]
self,
list_function: Callable[..., tuple[list[T], Meta]],
*args,
**kwargs,
) -> T | None:
entities, _ = list_function(*args, **kwargs)
return entities[0] if entities else None

Expand All @@ -69,15 +79,18 @@ def __init__(self, client: Client):
super().__init__(client)


class BoundModelBase:
Domain = TypeVar("Domain", bound=BaseDomain)


class BoundModelBase(Generic[Domain]):
"""Bound Model Base"""

model: type[BaseDomain]
model: type[Domain]

def __init__(
self,
client: ResourceClientBase,
data: dict,
data: dict[str, Any],
complete: bool = True,
):
"""
Expand All @@ -90,7 +103,7 @@ def __init__(
"""
self._client = client
self.complete = complete
self.data_model = self.model.from_dict(data)
self.data_model: Domain = self.model.from_dict(data)

def __getattr__(self, name: str): # type: ignore[no-untyped-def]
"""Allow magical access to the properties of the model
Expand All @@ -103,9 +116,10 @@ def __getattr__(self, name: str): # type: ignore[no-untyped-def]
value = getattr(self.data_model, name)
return value

def _get_self(self) -> BoundModelBase:
def _get_self(self) -> BoundModelBase[Domain]:
assert hasattr(self._client, "get_by_id")
return self._client.get_by_id(self.data_model.id)
assert hasattr(self.data_model, "id")
return self._client.get_by_id(self.data_model.id) # type: ignore

def reload(self) -> None:
"""Reloads the model and tries to get all data from the API"""
Expand Down
8 changes: 4 additions & 4 deletions hcloud/core/domain.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,18 +7,18 @@


class BaseDomain:
__api_properties__: tuple
__api_properties__: tuple[str, ...]

@classmethod
def from_dict(cls, data: dict): # type: ignore[no-untyped-def]
def from_dict(cls, data: dict[str, Any]): # type: ignore[no-untyped-def]
"""
Build the domain object from the data dict.
"""
supported_data = {k: v for k, v in data.items() if k in cls.__api_properties__}
return cls(**supported_data)

def __repr__(self) -> str:
kwargs = [f"{key}={getattr(self, key)!r}" for key in self.__api_properties__] # type: ignore[var-annotated]
kwargs = [f"{key}={getattr(self, key)!r}" for key in self.__api_properties__]
return f"{self.__class__.__qualname__}({', '.join(kwargs)})"

def __eq__(self, other: Any) -> bool:
Expand Down Expand Up @@ -119,7 +119,7 @@ def __init__(self, pagination: Pagination | None = None):
self.pagination = pagination

@classmethod
def parse_meta(cls, response: dict) -> Meta:
def parse_meta(cls, response: dict[str, Any]) -> Meta:
"""
If present, extract the meta details from the response and return a meta object.
"""
Expand Down
4 changes: 2 additions & 2 deletions hcloud/datacenters/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,12 +8,12 @@
from .domain import Datacenter, DatacenterServerTypes


class BoundDatacenter(BoundModelBase, Datacenter):
class BoundDatacenter(BoundModelBase[Datacenter], Datacenter):
_client: DatacentersClient

model = Datacenter

def __init__(self, client: DatacentersClient, data: dict):
def __init__(self, client: DatacentersClient, data: dict[str, Any]):
location = data.get("location")
if location is not None:
data["location"] = BoundLocation(client._parent.locations, location)
Expand Down
9 changes: 7 additions & 2 deletions hcloud/firewalls/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,12 +17,17 @@
from .._client import Client


class BoundFirewall(BoundModelBase, Firewall):
class BoundFirewall(BoundModelBase[Firewall], Firewall):
_client: FirewallsClient

model = Firewall

def __init__(self, client: FirewallsClient, data: dict, complete: bool = True):
def __init__(
self,
client: FirewallsClient,
data: dict[str, Any],
complete: bool = True,
):
rules = data.get("rules", [])
if rules:
rules = [
Expand Down
9 changes: 7 additions & 2 deletions hcloud/floating_ips/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,12 +13,17 @@
from ..servers import BoundServer, Server


class BoundFloatingIP(BoundModelBase, FloatingIP):
class BoundFloatingIP(BoundModelBase[FloatingIP], FloatingIP):
_client: FloatingIPsClient

model = FloatingIP

def __init__(self, client: FloatingIPsClient, data: dict, complete: bool = True):
def __init__(
self,
client: FloatingIPsClient,
data: dict[str, Any],
complete: bool = True,
):
# pylint: disable=import-outside-toplevel
from ..servers import BoundServer

Expand Down
11 changes: 8 additions & 3 deletions hcloud/floating_ips/domain.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,13 @@
from __future__ import annotations

from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, TypedDict

from ..core import BaseDomain, DomainIdentityMixin

if TYPE_CHECKING:
from ..actions import BoundAction
from ..locations import BoundLocation
from ..rdns import DNSPtr
from ..servers import BoundServer
from .client import BoundFloatingIP

Expand Down Expand Up @@ -63,10 +64,10 @@ def __init__(
description: str | None = None,
ip: str | None = None,
server: BoundServer | None = None,
dns_ptr: list[dict] | None = None,
dns_ptr: list[DNSPtr] | None = None,
home_location: BoundLocation | None = None,
blocked: bool | None = None,
protection: dict | None = None,
protection: FloatingIPProtection | None = None,
labels: dict[str, str] | None = None,
created: str | None = None,
name: str | None = None,
Expand All @@ -85,6 +86,10 @@ def __init__(
self.name = name


class FloatingIPProtection(TypedDict):
delete: bool


class CreateFloatingIPResponse(BaseDomain):
"""Create Floating IP Response Domain

Expand Down
8 changes: 6 additions & 2 deletions hcloud/images/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,12 +11,16 @@
from .._client import Client


class BoundImage(BoundModelBase, Image):
class BoundImage(BoundModelBase[Image], Image):
_client: ImagesClient

model = Image

def __init__(self, client: ImagesClient, data: dict):
def __init__(
self,
client: ImagesClient,
data: dict[str, Any],
):
# pylint: disable=import-outside-toplevel
from ..servers import BoundServer

Expand Down
8 changes: 6 additions & 2 deletions hcloud/images/domain.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
from __future__ import annotations

from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, TypedDict

from ..core import BaseDomain, DomainIdentityMixin

Expand Down Expand Up @@ -87,7 +87,7 @@ def __init__(
architecture: str | None = None,
rapid_deploy: bool | None = None,
created_from: Server | BoundServer | None = None,
protection: dict | None = None,
protection: ImageProtection | None = None,
labels: dict[str, str] | None = None,
status: str | None = None,
):
Expand All @@ -110,6 +110,10 @@ def __init__(
self.status = status


class ImageProtection(TypedDict):
delete: bool


class CreateImageResponse(BaseDomain):
"""Create Image Response Domain

Expand Down
Loading