Skip to content

Commit

Permalink
feat: Operation with resource (#1)
Browse files Browse the repository at this point in the history 
* fix: renamed models YandexLockboxResponse -> Operation, YandexLockboxError -> YandexCloudError

* feat: added computed field `resource` to Operation for manipulate with modified Lockbox resource

* chore: version bump
  • Loading branch information
@akimrx
akimrx authored Mar 23, 2024
1 parent 5d730a9 commit b2dbc33
Show file tree
Hide file tree
Showing 6 changed files with 135 additions and 94 deletions.
8 changes: 5 additions & 3 deletions README.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,7 @@ from yc_lockbox import YandexLockboxClient, INewSecret, INewSecretPayloadEntry

lockbox = YandexLockboxClient("oauth_or_iam_token")

my_secret = lockbox.create_secret(
create_secret_operation = lockbox.create_secret(
INewSecret(
folder_id="b1xxxxxxxxxxxxxx",
name="my-secret",
Expand All @@ -103,8 +103,10 @@ my_secret = lockbox.create_secret(
)
)

if my_secret.done:
print(my_secret.metadata.get("secretId"))
if create_secret_operation.done:
new_secret = create_secret_operation.resource
print(new_secret.id)
new_secret.deactivate()
```


Expand Down
2 changes: 1 addition & 1 deletion yc_lockbox/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@
from yc_lockbox._lockbox import YandexLockboxClient
from yc_lockbox._models import Secret, INewSecretPayloadEntry, INewSecret, INewSecretVersion, IUpdateSecret

__version__ = "0.0.2"
__version__ = "0.1.0"
__author__ = "Akim Faskhutdinov"
__author_email__ = "akimstrong@yandex.ru"
__license__ = "MIT"
Expand Down
6 changes: 3 additions & 3 deletions yc_lockbox/_adapters.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from typing import Any

from yc_lockbox._abc import AbstractHTTPAdapter
from yc_lockbox._models import YandexLockboxError
from yc_lockbox._models import YandexCloudError
from yc_lockbox._types import T

logger = logging.getLogger(__name__)
Expand Down Expand Up @@ -78,7 +78,7 @@ def request(
if raise_for_status:
response.raise_for_status()

return __class__.parse_response(response, response_model=YandexLockboxError)
return __class__.parse_response(response, response_model=YandexCloudError)

return __class__.parse_response(response, response_model=response_model)

Expand Down Expand Up @@ -161,6 +161,6 @@ async def request(
if raise_for_status:
response.raise_for_status()

return await __class__.parse_response(response, response_model=YandexLockboxError)
return await __class__.parse_response(response, response_model=YandexCloudError)

return await __class__.parse_response(response, response_model=response_model)
104 changes: 56 additions & 48 deletions yc_lockbox/_lockbox.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,8 @@
SecretVersion,
SecretsList,
SecretVersionsList,
YandexLockboxResponse,
YandexLockboxError,
Operation,
YandexCloudError,
INewSecret,
IUpdateSecret,
INewSecretVersion,
Expand Down Expand Up @@ -141,134 +141,138 @@ def _seekable_response(
item.inject_client(self)
yield item

def activate_secret(
self, secret_id: str, raise_for_status: bool = True
) -> YandexLockboxResponse | YandexLockboxError:
def activate_secret(self, secret_id: str, raise_for_status: bool = True) -> Operation | YandexCloudError:
"""
Activates the specified secret.
:param secret_id: Secret indentifier.
:param raise_for_status: If set to ``False`` returns :class:`YandexLockboxError` instead throw exception.
:param raise_for_status: If set to ``False`` returns :class:`YandexCloudError` instead throw exception.
Defaults to ``True``.
"""
url = f"{self.lockbox_base_url}/secrets/{secret_id}:activate"
return self.adapter.request(
response = self.adapter.request(
"POST",
url,
headers=self.auth_headers,
response_model=YandexLockboxResponse,
response_model=Operation,
raise_for_status=raise_for_status,
)
response.inject_client(self)
return response

def add_secret_version(
self, secret_id: str, version: INewSecretVersion, raise_for_status: bool = True
) -> YandexLockboxResponse | YandexLockboxError:
) -> Operation | YandexCloudError:
"""
Adds new version based on a previous one.
:param secret_id: Secret indentifier.
:param version: A new version object.
:param raise_for_status: If set to ``False`` returns :class:`YandexLockboxError` instead throw exception.
:param raise_for_status: If set to ``False`` returns :class:`YandexCloudError` instead throw exception.
Defaults to ``True``.
"""
url = f"{self.lockbox_base_url}/secrets/{secret_id}:addVersion"
payload = version.model_dump_json(by_alias=True, exclude_none=True)
return self.adapter.request(
response = self.adapter.request(
"POST",
url,
headers=self.auth_headers,
data=payload,
response_model=YandexLockboxResponse,
response_model=Operation,
raise_for_status=raise_for_status,
)
response.inject_client(self)
return response

def create_secret(
self, secret: INewSecret, raise_for_status: bool = True
) -> YandexLockboxResponse | YandexLockboxError:
def create_secret(self, secret: INewSecret, raise_for_status: bool = True) -> Operation | YandexCloudError:
"""
Creates a secret in the specified folder.
:param secret: A new secret object.
:param raise_for_status: If set to ``False`` returns :class:`YandexLockboxError` instead throw exception.
:param raise_for_status: If set to ``False`` returns :class:`YandexCloudError` instead throw exception.
Defaults to ``True``.
"""
url = f"{self.lockbox_base_url}/secrets"
payload = secret.model_dump_json(by_alias=True, exclude_none=True)
return self.adapter.request(
response = self.adapter.request(
"POST",
url,
headers=self.auth_headers,
data=payload,
response_model=YandexLockboxResponse,
response_model=Operation,
raise_for_status=raise_for_status,
)
response.inject_client(self)
return response

def cancel_secret_version_destruction(
self, secret_id: str, version_id: str, raise_for_status: bool = True
) -> YandexLockboxResponse | YandexLockboxError:
) -> Operation | YandexCloudError:
"""
Cancels previously scheduled version destruction, if the version hasn't been destroyed yet.
:param secret_id: Secret indentifier.
:param version_id: Secret version id to cancel destruction.
:param raise_for_status: If set to ``False`` returns :class:`YandexLockboxError` instead throw exception.
:param raise_for_status: If set to ``False`` returns :class:`YandexCloudError` instead throw exception.
Defaults to ``True``.
"""
url = f"{self.lockbox_base_url}/secrets/{secret_id}:cancelVersionDestruction"
payload = {"versionId": version_id}
return self.adapter.request(
response = self.adapter.request(
"POST",
url,
headers=self.auth_headers,
json=payload,
response_model=YandexLockboxResponse,
response_model=Operation,
raise_for_status=raise_for_status,
)
response.inject_client(self)
return response

def deactivate_secret(
self, secret_id: str, raise_for_status: bool = True
) -> YandexLockboxResponse | YandexLockboxError:
def deactivate_secret(self, secret_id: str, raise_for_status: bool = True) -> Operation | YandexCloudError:
"""
Deactivate a secret.
:param secret_id: Secret indentifier.
:param raise_for_status: If set to ``False`` returns :class:`YandexLockboxError` instead throw exception.
:param raise_for_status: If set to ``False`` returns :class:`YandexCloudError` instead throw exception.
Defaults to ``True``.
"""
url = f"{self.lockbox_base_url}/secrets/{secret_id}:deactivate"
return self.adapter.request(
response = self.adapter.request(
"POST",
url,
headers=self.auth_headers,
response_model=YandexLockboxResponse,
response_model=Operation,
raise_for_status=raise_for_status,
)
response.inject_client(self)
return response

def delete_secret(
self, secret_id: str, raise_for_status: bool = True
) -> YandexLockboxResponse | YandexLockboxError:
def delete_secret(self, secret_id: str, raise_for_status: bool = True) -> Operation | YandexCloudError:
"""
Deletes the specified secret.
:param secret_id: Secret indentifier.
:param raise_for_status: If set to ``False`` returns :class:`YandexLockboxError` instead throw exception.
:param raise_for_status: If set to ``False`` returns :class:`YandexCloudError` instead throw exception.
Defaults to ``True``.
"""
url = f"{self.lockbox_base_url}/secrets/{secret_id}"
return self.adapter.request(
response = self.adapter.request(
"DELETE",
url,
headers=self.auth_headers,
response_model=YandexLockboxResponse,
response_model=Operation,
raise_for_status=raise_for_status,
)
response.inject_client(self)
return response

def get_secret(self, secret_id: str, raise_for_status: bool = True) -> Secret | YandexLockboxError:
def get_secret(self, secret_id: str, raise_for_status: bool = True) -> Secret | YandexCloudError:
"""
Get lockbox secret by ID.
:param secret_id: Secret identifier.
:param raise_for_status: If set to ``False`` returns :class:`YandexLockboxError` instead throw exception.
:param raise_for_status: If set to ``False`` returns :class:`YandexCloudError` instead throw exception.
Defaults to ``True``.
"""
url = f"{self.lockbox_base_url}/secrets/{secret_id}"
Expand All @@ -287,13 +291,13 @@ def get_secret_payload(
secret_id: str,
version_id: str | None = None,
raise_for_status: bool = True,
) -> SecretPayload | YandexLockboxError:
) -> SecretPayload | YandexCloudError:
"""
Get lockbox secret payload by ID and optional version.
:param secret_id: Secret identifier.
:param version_id: Secret version. Optional.
:param raise_for_status: If set to ``False`` returns :class:`YandexLockboxError` instead throw exception.
:param raise_for_status: If set to ``False`` returns :class:`YandexCloudError` instead throw exception.
Defaults to ``True``.
"""
url = f"{self.payload_lockbox_base_url}/secrets/{secret_id}/payload"
Expand All @@ -314,7 +318,7 @@ def list_secrets(
page_token: str | None = None,
raise_for_status: bool = True,
iterator: bool = False,
) -> SecretsList | Iterator[Secret] | YandexLockboxError:
) -> SecretsList | Iterator[Secret] | YandexCloudError:
"""
Retrieves the list of secrets in the specified folder.
Expand Down Expand Up @@ -362,7 +366,7 @@ def list_secret_versions(
page_token: str | None = None,
raise_for_status: bool = True,
iterator: bool = False,
) -> SecretVersionsList | Iterator[SecretVersion] | YandexLockboxError:
) -> SecretVersionsList | Iterator[SecretVersion] | YandexCloudError:
"""
Retrieves the list of versions of the specified secret.
Expand Down Expand Up @@ -397,7 +401,7 @@ def list_secret_versions(

def schedule_secret_version_destruction(
self, secret_id: str, version_id: str, pending_period: int = 604800, raise_for_status: bool = True
) -> YandexLockboxResponse | YandexLockboxError:
) -> Operation | YandexCloudError:
"""
Schedules the specified version for destruction.
Scheduled destruction can be cancelled with the :func:`cancel_secret_version_destruction()` method.
Expand All @@ -406,7 +410,7 @@ def schedule_secret_version_destruction(
:param version_id: ID of the version to be destroyed.
:param pending_period: Time interval in seconds between the version destruction request and actual destruction.
Default value: ``604800`` (i.e. 7 days).
:param raise_for_status: If set to ``False`` returns :class:`YandexLockboxError` instead throw exception.
:param raise_for_status: If set to ``False`` returns :class:`YandexCloudError` instead throw exception.
Defaults to ``True``.
"""
if isinstance(pending_period, int):
Expand All @@ -420,22 +424,24 @@ def schedule_secret_version_destruction(

url = f"{self.lockbox_base_url}/secrets/{secret_id}:scheduleVersionDestruction"
payload = {"versionId": version_id, "pendingPeriod": pending_period}
return self.adapter.request(
response = self.adapter.request(
"POST",
url,
headers=self.auth_headers,
json=payload,
response_model=YandexLockboxResponse,
response_model=Operation,
raise_for_status=raise_for_status,
)
response.inject_client(self)
return response

# TODO: implement
def set_secret_access_bindings(self, *args, **kwargs):
raise NotImplementedError

def update_secret(
self, secret_id: str, data: IUpdateSecret, raise_for_status: bool = True
) -> YandexLockboxResponse | YandexLockboxError:
) -> Operation | YandexCloudError:
"""
Updates the specified secret.
Expand All @@ -448,19 +454,21 @@ def update_secret(
The default value for most fields is null or 0.
If ``updateMask`` is not sent in the request, all fields values will be updated.
Fields specified in the request will be updated to provided values. The rest of the fields will be reset to the default.
:param raise_for_status: If set to ``False`` returns :class:`YandexLockboxError` instead throw exception.
:param raise_for_status: If set to ``False`` returns :class:`YandexCloudError` instead throw exception.
Defaults to ``True``.
"""
url = f"{self.lockbox_base_url}/secrets/{secret_id}"
payload = data.model_dump_json(by_alias=True)
return self.adapter.request(
response = self.adapter.request(
"PATCH",
url,
headers=self.auth_headers,
data=payload,
response_model=YandexLockboxResponse,
response_model=Operation,
raise_for_status=raise_for_status,
)
response.inject_client(self)
return response

# TODO: implement
def update_secret_access_bindings(self, *args, **kwargs):
Expand Down
Loading

0 comments on commit b2dbc33

Please sign in to comment.