​

prefect_aws.secrets_manager

​

Functions

​

read_secret

read_secret(secret_name: str, aws_credentials: AwsCredentials, version_id: Optional[str] = None, version_stage: Optional[str] = None) -> Union[str, bytes]

• secret_name: Name of stored secret.
• aws_credentials: Credentials to use for authentication with AWS.
• version_id: Specifies version of secret to read. Defaults to the most recent version if not given.
• version_stage: Specifies the version stage of the secret to read. Defaults to AWS_CURRENT if not given.

• The secret values as a str or bytes depending on the format in which the secret was stored.

​

update_secret

update_secret(secret_name: str, secret_value: Union[str, bytes], aws_credentials: AwsCredentials, description: Optional[str] = None) -> Dict[str, str]

• secret_name: Name of secret to update.
• secret_value: Desired value of the secret. Can be either str or bytes.
• aws_credentials: Credentials to use for authentication with AWS.
• description: Desired description of the secret.

• A dict containing the secret ARN (Amazon Resource Name), name, and current version ID.

{
    "ARN": str,
    "Name": str,
    "VersionId": str
}

​

create_secret

create_secret(secret_name: str, secret_value: Union[str, bytes], aws_credentials: AwsCredentials, description: Optional[str] = None, tags: Optional[List[Dict[str, str]]] = None) -> Dict[str, str]

• secret_name: The name of the secret to create.
• secret_value: The value to store in the created secret.
• aws_credentials: Credentials to use for authentication with AWS.
• description: A description for the created secret.
• tags: A list of tags to attach to the secret. Each tag should be specified as a dictionary in the following format:

{
    "Key"\: str,
    "Value"\: str
}

• A dict containing the secret ARN (Amazon Resource Name), name, and current version ID.

{
    "ARN": str,
    "Name": str,
    "VersionId": str
}

from prefect import flow
from prefect_aws import AwsCredentials
from prefect_aws.secrets_manager import create_secret

@flow
def example_create_secret():
    aws_credentials = AwsCredentials(
        aws_access_key_id="access_key_id",
        aws_secret_access_key="secret_access_key"
    )
    create_secret(
        secret_name="life_the_universe_and_everything",
        secret_value="42",
        aws_credentials=aws_credentials
    )

example_create_secret()

​

delete_secret

delete_secret(secret_name: str, aws_credentials: AwsCredentials, recovery_window_in_days: int = 30, force_delete_without_recovery: bool = False) -> Dict[str, str]

• secret_name: Name of the secret to be deleted.
• aws_credentials: Credentials to use for authentication with AWS.
• recovery_window_in_days: Number of days a secret should be recoverable for before permanent deletion. Minimum window is 7 days and maximum window is 30 days. If force_delete_without_recovery is set to True, this value will be ignored.
• force_delete_without_recovery: If True, the secret will be immediately deleted and will not be recoverable.

• A dict containing the secret ARN (Amazon Resource Name), name, and deletion date of the secret. DeletionDate is the date and time of the delete request plus the number of days in recovery_window_in_days.

{
    "ARN": str,
    "Name": str,
    "DeletionDate": datetime.datetime
}

from prefect import flow
from prefect_aws import AwsCredentials
from prefect_aws.secrets_manager import delete_secret

@flow
def example_delete_secret_immediately():
    aws_credentials = AwsCredentials(
        aws_access_key_id="access_key_id",
        aws_secret_access_key="secret_access_key"
    )
    delete_secret(
        secret_name="life_the_universe_and_everything",
        aws_credentials=aws_credentials,
        force_delete_without_recovery: True
    )

example_delete_secret_immediately()

from prefect import flow
from prefect_aws import AwsCredentials
from prefect_aws.secrets_manager import delete_secret

@flow
def example_delete_secret_with_recovery_window():
    aws_credentials = AwsCredentials(
        aws_access_key_id="access_key_id",
        aws_secret_access_key="secret_access_key"
    )
    delete_secret(
        secret_name="life_the_universe_and_everything",
        aws_credentials=aws_credentials,
        recovery_window_in_days=90
    )

example_delete_secret_with_recovery_window()

​

Classes

​

AwsSecret

• aws_credentials: The credentials to use for authentication with AWS.
• secret_name: The name of the secret.

​

adelete_secret

adelete_secret(self, recovery_window_in_days: int = 30, force_delete_without_recovery: bool = False, **delete_kwargs: Dict[str, Any]) -> str

• recovery_window_in_days: The number of days to wait before permanently deleting the secret. Must be between 7 and 30 days.
• force_delete_without_recovery: If True, the secret will be deleted immediately without a recovery window.
• **delete_kwargs: Additional keyword arguments to pass to the delete_secret method of the boto3 client.

• The path that the secret was deleted from.

secrets_manager = SecretsManager.load("MY_BLOCK")
await secrets_manager.adelete_secret(recovery_window_in_days=15)

​

aread_secret

aread_secret(self, version_id: Optional[str] = None, version_stage: Optional[str] = None, **read_kwargs: Any) -> bytes

• version_id: The version of the secret to read. If not provided, the latest version will be read.
• version_stage: The version stage of the secret to read. If not provided, the latest version will be read.
• read_kwargs: Additional keyword arguments to pass to the get_secret_value method of the boto3 client.

• The secret data.

secrets_manager = SecretsManager.load("MY_BLOCK")
await secrets_manager.aread_secret()

​

awrite_secret

awrite_secret(self, secret_data: bytes, **put_or_create_secret_kwargs: Dict[str, Any]) -> str

• secret_data: The secret data to write.
• **put_or_create_secret_kwargs: Additional keyword arguments to pass to put_secret_value or create_secret method of the boto3 client.

• The path that the secret was written to.

secrets_manager = SecretsManager.load("MY_BLOCK")
await secrets_manager.awrite_secret(b"my_secret_data")

​

delete_secret

delete_secret(self, recovery_window_in_days: int = 30, force_delete_without_recovery: bool = False, **delete_kwargs: Dict[str, Any]) -> str

• recovery_window_in_days: The number of days to wait before permanently deleting the secret. Must be between 7 and 30 days.
• force_delete_without_recovery: If True, the secret will be deleted immediately without a recovery window.
• **delete_kwargs: Additional keyword arguments to pass to the delete_secret method of the boto3 client.

• The path that the secret was deleted from.

secrets_manager = SecretsManager.load("MY_BLOCK")
secrets_manager.delete_secret(recovery_window_in_days=15)

​

read_secret

read_secret(self, version_id: Optional[str] = None, version_stage: Optional[str] = None, **read_kwargs: Any) -> bytes

• version_id: The version of the secret to read. If not provided, the latest version will be read.
• version_stage: The version stage of the secret to read. If not provided, the latest version will be read.
• read_kwargs: Additional keyword arguments to pass to the get_secret_value method of the boto3 client.

• The secret data.

secrets_manager = SecretsManager.load("MY_BLOCK")
secrets_manager.read_secret()

​

write_secret

write_secret(self, secret_data: bytes, **put_or_create_secret_kwargs: Dict[str, Any]) -> str

• secret_data: The secret data to write.
• **put_or_create_secret_kwargs: Additional keyword arguments to pass to put_secret_value or create_secret method of the boto3 client.

• The path that the secret was written to.

secrets_manager = SecretsManager.load("MY_BLOCK")
secrets_manager.write_secret(b"my_secret_data")