-
Notifications
You must be signed in to change notification settings - Fork 51
Expand file tree
/
Copy pathclient.py
More file actions
140 lines (111 loc) · 4.03 KB
/
client.py
File metadata and controls
140 lines (111 loc) · 4.03 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
from __future__ import annotations
import warnings
from collections.abc import Callable
from typing import TYPE_CHECKING, Any, ClassVar, Generic, TypeVar
from .domain import BaseDomain
if TYPE_CHECKING:
from .._client import Client, ClientBase
from .domain import Meta
T = TypeVar("T")
class ResourceClientBase:
_base_url: ClassVar[str]
_parent: Client
_client: ClientBase
max_per_page: int = 50
def __init__(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,
list_function: Callable[..., tuple[list[T], Meta]],
*args,
**kwargs,
) -> list[T]:
results = []
page = 1
while page:
# The *PageResult tuples MUST have the following structure
# `(result: List[Bound*], meta: Meta)`
result, meta = list_function(
*args, page=page, per_page=self.max_per_page, **kwargs
)
if result:
results.extend(result)
if meta and meta.pagination and meta.pagination.next_page:
page = meta.pagination.next_page
else:
page = 0
return results
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
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)
Domain = TypeVar("Domain", bound=BaseDomain)
class BoundModelBase(Generic[Domain]):
"""Bound Model Base"""
model: type[Domain]
def __init__(
self,
client: ResourceClientBase,
data: dict[str, Any],
complete: bool = True,
):
"""
:param client:
The client for the specific model to use
:param data:
The data of the model
:param complete: bool
False if not all attributes of the model fetched
"""
self._client = client
self.complete = complete
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
:param name: str
:return:
"""
value = getattr(self.data_model, name)
if not value and not self.complete:
self.reload()
value = getattr(self.data_model, name)
return value
def _get_self(self) -> BoundModelBase[Domain]:
assert hasattr(self._client, "get_by_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"""
bound_model = self._get_self()
self.data_model = bound_model.data_model
self.complete = True
def __repr__(self) -> str:
# Override and reset hcloud.core.domain.BaseDomain.__repr__ method for bound
# models, as they will generate a lot of API call trying to print all the fields
# of the model.
return object.__repr__(self)
def __eq__(self, other: Any) -> bool:
"""Compare a bound model object with another of the same type."""
if not isinstance(other, self.__class__):
return NotImplemented
return self.data_model == other.data_model