You've already forked keep-it-secret
v1.1.0
This commit is contained in:
@@ -6,7 +6,7 @@ from .fields import ( # noqa: F401
|
||||
)
|
||||
from .secrets import Secrets # noqa: F401
|
||||
|
||||
__version__ = '1.0.0'
|
||||
__version__ = '1.1.0'
|
||||
|
||||
__all__ = [
|
||||
'AbstractField',
|
||||
|
||||
@@ -57,6 +57,11 @@ class AWSSecrets(Secrets):
|
||||
:type: ``str | None``
|
||||
"""
|
||||
|
||||
def __init__(self, parent: Secrets | None = None):
|
||||
super().__init__(parent=parent)
|
||||
|
||||
self.client: PAWSSecretsManagerClient | None = None
|
||||
|
||||
def as_boto3_client_kwargs(self) -> dict[str, typing.Any]:
|
||||
"""
|
||||
Return representation of the mapped variables for use in
|
||||
@@ -78,6 +83,15 @@ class AWSSecrets(Secrets):
|
||||
|
||||
return result
|
||||
|
||||
def get_client(self) -> PAWSSecretsManagerClient:
|
||||
if self.client is None:
|
||||
self.client = boto3.client(
|
||||
'secretsmanager',
|
||||
**self.as_boto3_client_kwargs(),
|
||||
)
|
||||
|
||||
return self.client
|
||||
|
||||
|
||||
class AWSSecretsManagerField(Field):
|
||||
"""
|
||||
@@ -87,7 +101,7 @@ class AWSSecretsManagerField(Field):
|
||||
:param secret_id: ID of the secret to fetch.
|
||||
:param default: Default value. Defaults to ``None``.
|
||||
:param decoder: A callable to decode the fetched value. Defaults to
|
||||
``json.loads``.
|
||||
:py:func:`json.loads`.
|
||||
"""
|
||||
def __init__(self,
|
||||
secret_id: str,
|
||||
@@ -100,8 +114,6 @@ class AWSSecretsManagerField(Field):
|
||||
self.default = default
|
||||
self.decoder = decoder
|
||||
|
||||
self.client: PAWSSecretsManagerClient | None = None
|
||||
|
||||
@classmethod
|
||||
def new(cls: type[AWSSecretsManagerField], # type: ignore[override]
|
||||
secret_id: str,
|
||||
@@ -110,33 +122,25 @@ class AWSSecretsManagerField(Field):
|
||||
**field_options) -> AWSSecretsManagerField:
|
||||
return cls(secret_id, default=default, decoder=decoder, **field_options)
|
||||
|
||||
def get_client(self, secrets: Secrets) -> PAWSSecretsManagerClient:
|
||||
if self.client is None:
|
||||
aws_secrets = typing.cast(AWSSecrets, secrets.aws) # type: ignore[attr-defined]
|
||||
|
||||
self.client = boto3.client(
|
||||
'secretsmanager',
|
||||
**aws_secrets.as_boto3_client_kwargs(),
|
||||
)
|
||||
|
||||
return self.client
|
||||
|
||||
def get_value(self, secrets: Secrets) -> typing.Any:
|
||||
"""
|
||||
Retrieve, decode and return the secret specified by *secret_id*.
|
||||
|
||||
Depends on :py:class:`AWSSecrets` to be declared in ``secrets.aws``
|
||||
field.
|
||||
Depends on :py:class:`AWSSecrets` to be declared in ``aws`` field on
|
||||
``secrets`` or one of its parents.
|
||||
|
||||
:raises DependencyMissing: Signal that ``secrets.aws`` field is
|
||||
missing.
|
||||
:raises RequiredValueMissing: Signal the field's value is required but
|
||||
*secret_id* is not present in the Secrets Manager.
|
||||
"""
|
||||
if hasattr(secrets, 'aws') is False:
|
||||
aws_secrets: AWSSecrets = secrets.resolve_dependency(
|
||||
'aws', include_parents=True,
|
||||
)
|
||||
if aws_secrets is secrets.UNRESOLVED_DEPENDENCY:
|
||||
raise self.DependencyMissing('aws')
|
||||
|
||||
client = self.get_client(secrets)
|
||||
client = aws_secrets.get_client()
|
||||
|
||||
try:
|
||||
secret = client.get_secret_value(SecretId=self.secret_id)
|
||||
|
||||
@@ -95,6 +95,10 @@ class Secrets(metaclass=SecretsBase):
|
||||
|
||||
:param parent: The parent :py:class:`Secrets` subclass or ``None``.
|
||||
"""
|
||||
|
||||
#: Sentinel for unresolved dependency.
|
||||
UNRESOLVED_DEPENDENCY: list[typing.Any] = []
|
||||
|
||||
__secrets_fields__: TFields
|
||||
|
||||
def __init__(self, parent: Secrets | None = None):
|
||||
@@ -104,3 +108,24 @@ class Secrets(metaclass=SecretsBase):
|
||||
|
||||
for field_name, field in self.__secrets_fields__.items():
|
||||
self.__secrets_data__[field_name] = getattr(self, field_name)
|
||||
|
||||
def resolve_dependency(self,
|
||||
name: str,
|
||||
*,
|
||||
include_parents: bool = True) -> typing.Any:
|
||||
"""
|
||||
Resolve a dependency field identified by *name* and return its value.
|
||||
returns :py:attr:`keep_it_secret.Secrets.UNRESOLVED_DEPENDENCY` if
|
||||
the value can't be resolved.
|
||||
|
||||
:param include_parents: Recursively include parents, if any.
|
||||
"""
|
||||
if name in self.__secrets_fields__:
|
||||
return getattr(self, name)
|
||||
|
||||
if include_parents is True and self.__secrets_parent__ is not None:
|
||||
return self.__secrets_parent__.resolve_dependency(
|
||||
name, include_parents=include_parents,
|
||||
)
|
||||
|
||||
return self.UNRESOLVED_DEPENDENCY
|
||||
|
||||
Reference in New Issue
Block a user