File: client.py

package info (click to toggle)
hcloud-python 2.3.0-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 1,396 kB
  • sloc: python: 15,311; makefile: 43; javascript: 3
file content (371 lines) | stat: -rw-r--r-- 15,484 bytes parent folder | download
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
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
from __future__ import annotations

from typing import TYPE_CHECKING, Any, NamedTuple

from ..actions import ActionsPageResult, BoundAction, ResourceActionsClient
from ..core import BoundModelBase, ClientEntityBase, Meta
from .domain import (
    Certificate,
    CreateManagedCertificateResponse,
    ManagedCertificateError,
    ManagedCertificateStatus,
)

if TYPE_CHECKING:
    from .._client import Client


class BoundCertificate(BoundModelBase, Certificate):
    _client: CertificatesClient

    model = Certificate

    def __init__(self, client: CertificatesClient, data: dict, complete: bool = True):
        status = data.get("status")
        if status is not None:
            error_data = status.get("error")
            error = None
            if error_data:
                error = ManagedCertificateError(
                    code=error_data["code"], message=error_data["message"]
                )
            data["status"] = ManagedCertificateStatus(
                issuance=status["issuance"], renewal=status["renewal"], error=error
            )
        super().__init__(client, data, complete)

    def get_actions_list(
        self,
        status: list[str] | None = None,
        sort: list[str] | None = None,
        page: int | None = None,
        per_page: int | None = None,
    ) -> ActionsPageResult:
        """Returns all action objects for a Certificate.

        :param status: List[str] (optional)
               Response will have only actions with specified statuses. Choices: `running` `success` `error`
        :param sort: List[str] (optional)
               Specify how the results are sorted. Choices: `id` `id:asc` `id:desc` `command` `command:asc` `command:desc` `status` `status:asc` `status:desc` `progress` `progress:asc` `progress:desc` `started` `started:asc` `started:desc` `finished` `finished:asc` `finished:desc`
        :param page: int (optional)
               Specifies the page to fetch
        :param per_page: int (optional)
               Specifies how many results are returned by page
        :return: (List[:class:`BoundAction <hcloud.actions.client.BoundAction>`], :class:`Meta <hcloud.core.domain.Meta>`)
        """
        return self._client.get_actions_list(self, status, sort, page, per_page)

    def get_actions(
        self,
        status: list[str] | None = None,
        sort: list[str] | None = None,
    ) -> list[BoundAction]:
        """Returns all action objects for a Certificate.

        :param status: List[str] (optional)
               Response will have only actions with specified statuses. Choices: `running` `success` `error`
        :param sort: List[str] (optional)
               Specify how the results are sorted. Choices: `id` `id:asc` `id:desc` `command` `command:asc` `command:desc` `status` `status:asc` `status:desc` `progress` `progress:asc` `progress:desc` `started` `started:asc` `started:desc` `finished` `finished:asc` `finished:desc`
        :return: List[:class:`BoundAction <hcloud.actions.client.BoundAction>`]
        """
        return self._client.get_actions(self, status, sort)

    def update(
        self,
        name: str | None = None,
        labels: dict[str, str] | None = None,
    ) -> BoundCertificate:
        """Updates an certificate. You can update an certificate name and the certificate labels.

        :param name: str (optional)
               New name to set
        :param labels: Dict[str, str] (optional)
               User-defined labels (key-value pairs)
        :return: :class:`BoundCertificate <hcloud.certificates.client.BoundCertificate>`
        """
        return self._client.update(self, name, labels)

    def delete(self) -> bool:
        """Deletes a certificate.
        :return: boolean
        """
        return self._client.delete(self)

    def retry_issuance(self) -> BoundAction:
        """Retry a failed Certificate issuance or renewal.
        :return: BoundAction
        """
        return self._client.retry_issuance(self)


class CertificatesPageResult(NamedTuple):
    certificates: list[BoundCertificate]
    meta: Meta | None


class CertificatesClient(ClientEntityBase):
    _client: Client

    actions: ResourceActionsClient
    """Certificates scoped actions client

    :type: :class:`ResourceActionsClient <hcloud.actions.client.ResourceActionsClient>`
    """

    def __init__(self, client: Client):
        super().__init__(client)
        self.actions = ResourceActionsClient(client, "/certificates")

    def get_by_id(self, id: int) -> BoundCertificate:
        """Get a specific certificate by its ID.

        :param id: int
        :return: :class:`BoundCertificate <hcloud.certificates.client.BoundCertificate>`
        """
        response = self._client.request(url=f"/certificates/{id}", method="GET")
        return BoundCertificate(self, response["certificate"])

    def get_list(
        self,
        name: str | None = None,
        label_selector: str | None = None,
        page: int | None = None,
        per_page: int | None = None,
    ) -> CertificatesPageResult:
        """Get a list of certificates

        :param name: str (optional)
               Can be used to filter certificates by their name.
        :param label_selector: str (optional)
               Can be used to filter certificates by labels. The response will only contain certificates matching the label selector.
        :param page: int (optional)
               Specifies the page to fetch
        :param per_page: int (optional)
               Specifies how many results are returned by page
        :return: (List[:class:`BoundCertificate <hcloud.certificates.client.BoundCertificate>`], :class:`Meta <hcloud.core.domain.Meta>`)
        """
        params: dict[str, Any] = {}
        if name is not None:
            params["name"] = name

        if label_selector is not None:
            params["label_selector"] = label_selector

        if page is not None:
            params["page"] = page

        if per_page is not None:
            params["per_page"] = per_page

        response = self._client.request(
            url="/certificates", method="GET", params=params
        )

        certificates = [
            BoundCertificate(self, certificate_data)
            for certificate_data in response["certificates"]
        ]

        return CertificatesPageResult(certificates, Meta.parse_meta(response))

    def get_all(
        self,
        name: str | None = None,
        label_selector: str | None = None,
    ) -> list[BoundCertificate]:
        """Get all certificates

        :param name: str (optional)
               Can be used to filter certificates by their name.
        :param label_selector: str (optional)
               Can be used to filter certificates by labels. The response will only contain certificates matching the label selector.
        :return: List[:class:`BoundCertificate <hcloud.certificates.client.BoundCertificate>`]
        """
        return self._iter_pages(self.get_list, name=name, label_selector=label_selector)

    def get_by_name(self, name: str) -> BoundCertificate | None:
        """Get certificate by name

        :param name: str
               Used to get certificate by name.
        :return: :class:`BoundCertificate <hcloud.certificates.client.BoundCertificate>`
        """
        return self._get_first_by(name=name)

    def create(
        self,
        name: str,
        certificate: str,
        private_key: str,
        labels: dict[str, str] | None = None,
    ) -> BoundCertificate:
        """Creates a new Certificate with the given name, certificate and private_key. This methods allows only creating
           custom uploaded certificates. If you want to create a managed certificate use :func:`~hcloud.certificates.client.CertificatesClient.create_managed`

        :param name: str
        :param certificate: str
               Certificate and chain in PEM format, in order so that each record directly certifies the one preceding
        :param private_key: str
               Certificate key in PEM format
        :param labels: Dict[str, str] (optional)
               User-defined labels (key-value pairs)
        :return: :class:`BoundCertificate <hcloud.certificates.client.BoundCertificate>`
        """
        data: dict[str, Any] = {
            "name": name,
            "certificate": certificate,
            "private_key": private_key,
            "type": Certificate.TYPE_UPLOADED,
        }
        if labels is not None:
            data["labels"] = labels
        response = self._client.request(url="/certificates", method="POST", json=data)
        return BoundCertificate(self, response["certificate"])

    def create_managed(
        self,
        name: str,
        domain_names: list[str],
        labels: dict[str, str] | None = None,
    ) -> CreateManagedCertificateResponse:
        """Creates a new managed Certificate with the given name and domain names. This methods allows only creating
           managed certificates for domains that are using the Hetzner DNS service. If you want to create a custom uploaded certificate use :func:`~hcloud.certificates.client.CertificatesClient.create`

        :param name: str
        :param domain_names: List[str]
               Domains and subdomains that should be contained in the Certificate
        :param labels: Dict[str, str] (optional)
               User-defined labels (key-value pairs)
        :return: :class:`BoundCertificate <hcloud.certificates.client.BoundCertificate>`
        """
        data: dict[str, Any] = {
            "name": name,
            "type": Certificate.TYPE_MANAGED,
            "domain_names": domain_names,
        }
        if labels is not None:
            data["labels"] = labels
        response = self._client.request(url="/certificates", method="POST", json=data)
        return CreateManagedCertificateResponse(
            certificate=BoundCertificate(self, response["certificate"]),
            action=BoundAction(self._client.actions, response["action"]),
        )

    def update(
        self,
        certificate: Certificate | BoundCertificate,
        name: str | None = None,
        labels: dict[str, str] | None = None,
    ) -> BoundCertificate:
        """Updates a Certificate. You can update a certificate name and labels.

        :param certificate: :class:`BoundCertificate <hcloud.certificates.client.BoundCertificate>` or  :class:`Certificate <hcloud.certificates.domain.Certificate>`
        :param name: str (optional)
               New name to set
        :param labels: Dict[str, str] (optional)
               User-defined labels (key-value pairs)
        :return: :class:`BoundCertificate <hcloud.certificates.client.BoundCertificate>`
        """
        data: dict[str, Any] = {}
        if name is not None:
            data["name"] = name
        if labels is not None:
            data["labels"] = labels
        response = self._client.request(
            url=f"/certificates/{certificate.id}",
            method="PUT",
            json=data,
        )
        return BoundCertificate(self, response["certificate"])

    def delete(self, certificate: Certificate | BoundCertificate) -> bool:
        """Deletes a certificate.

        :param certificate: :class:`BoundCertificate <hcloud.certificates.client.BoundCertificate>` or  :class:`Certificate <hcloud.certificates.domain.Certificate>`
        :return: True
        """
        self._client.request(
            url=f"/certificates/{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
        return True

    def get_actions_list(
        self,
        certificate: Certificate | BoundCertificate,
        status: list[str] | None = None,
        sort: list[str] | None = None,
        page: int | None = None,
        per_page: int | None = None,
    ) -> ActionsPageResult:
        """Returns all action objects for a Certificate.

        :param certificate: :class:`BoundCertificate <hcloud.certificates.client.BoundCertificate>` or :class:`Certificate <hcloud.certificates.domain.Certificate>`
        :param status: List[str] (optional)
               Response will have only actions with specified statuses. Choices: `running` `success` `error`
        :param sort: List[str] (optional)
               Specify how the results are sorted. Choices: `id` `id:asc` `id:desc` `command` `command:asc` `command:desc` `status` `status:asc` `status:desc` `progress` `progress:asc` `progress:desc` `started` `started:asc` `started:desc` `finished` `finished:asc` `finished:desc`
        :param page: int (optional)
               Specifies the page to fetch
        :param per_page: int (optional)
               Specifies how many results are returned by page
        :return: (List[:class:`BoundAction <hcloud.actions.client.BoundAction>`], :class:`Meta <hcloud.core.domain.Meta>`)
        """
        params: dict[str, Any] = {}
        if status is not None:
            params["status"] = status
        if sort is not None:
            params["sort"] = sort
        if page is not None:
            params["page"] = page
        if per_page is not None:
            params["per_page"] = per_page

        response = self._client.request(
            url=f"/certificates/{certificate.id}/actions",
            method="GET",
            params=params,
        )
        actions = [
            BoundAction(self._client.actions, action_data)
            for action_data in response["actions"]
        ]
        return ActionsPageResult(actions, Meta.parse_meta(response))

    def get_actions(
        self,
        certificate: Certificate | BoundCertificate,
        status: list[str] | None = None,
        sort: list[str] | None = None,
    ) -> list[BoundAction]:
        """Returns all action objects for a Certificate.

        :param certificate: :class:`BoundCertificate <hcloud.certificates.client.BoundCertificate>` or :class:`Certificate <hcloud.certificates.domain.Certificate>`
        :param status: List[str] (optional)
               Response will have only actions with specified statuses. Choices: `running` `success` `error`
        :param sort: List[str] (optional)
               Specify how the results are sorted. Choices: `id` `id:asc` `id:desc` `command` `command:asc` `command:desc` `status` `status:asc` `status:desc` `progress` `progress:asc` `progress:desc` `started` `started:asc` `started:desc` `finished` `finished:asc` `finished:desc`
        :return: List[:class:`BoundAction <hcloud.actions.client.BoundAction>`]
        """
        return self._iter_pages(
            self.get_actions_list,
            certificate,
            status=status,
            sort=sort,
        )

    def retry_issuance(
        self,
        certificate: Certificate | BoundCertificate,
    ) -> BoundAction:
        """Returns all action objects for a Certificate.

        :param certificate: :class:`BoundCertificate <hcloud.certificates.client.BoundCertificate>` or :class:`Certificate <hcloud.certificates.domain.Certificate>`
        :return: :class:`BoundAction <hcloud.actions.client.BoundAction>`
        """
        response = self._client.request(
            url=f"/certificates/{certificate.id}/actions/retry",
            method="POST",
        )
        return BoundAction(self._client.actions, response["action"])