prefect_aws.secrets_manager

Tasks for interacting with AWS Secrets Manager

Classes

AwsSecret (SecretBlock) pydantic-model

Manages a secret in AWS's Secrets Manager.

Attributes:

Name	Type	Description
aws_credentials	AwsCredentials	The credentials to use for authentication with AWS.
secret_name	str	The name of the secret.

Source code in prefect_aws/secrets_manager.py

class AwsSecret(SecretBlock):
    """
    Manages a secret in AWS's Secrets Manager.

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

    _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/d74b16fe84ce626345adf235a47008fea2869a60-225x225.png"  # noqa
    _block_type_name = "AWS Secret"
    _documentation_url = "https://prefecthq.github.io/prefect-aws/secrets_manager/#prefect_aws.secrets_manager.AwsSecret"  # noqa

    aws_credentials: AwsCredentials
    secret_name: str = Field(default=..., description="The name of the secret.")

    @sync_compatible
    async def read_secret(
        self,
        version_id: str = None,
        version_stage: str = None,
        **read_kwargs: Dict[str, Any],
    ) -> bytes:
        """
        Reads the secret from the secret storage service.

        Args:
            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.

        Returns:
            The secret data.

        Examples:
            Reads a secret.
            ```python
            secrets_manager = SecretsManager.load("MY_BLOCK")
            secrets_manager.read_secret()
            ```
        """
        client = self.aws_credentials.get_secrets_manager_client()
        if version_id is not None:
            read_kwargs["VersionId"] = version_id
        if version_stage is not None:
            read_kwargs["VersionStage"] = version_stage
        response = await run_sync_in_worker_thread(
            client.get_secret_value, SecretId=self.secret_name, **read_kwargs
        )
        if "SecretBinary" in response:
            secret = response["SecretBinary"]
        elif "SecretString" in response:
            secret = response["SecretString"]
        arn = response["ARN"]
        self.logger.info(f"The secret {arn!r} data was successfully read.")
        return secret

    @sync_compatible
    async def write_secret(
        self, secret_data: bytes, **put_or_create_secret_kwargs: Dict[str, Any]
    ) -> str:
        """
        Writes the secret to the secret storage service as a SecretBinary;
        if it doesn't exist, it will be created.

        Args:
            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.

        Returns:
            The path that the secret was written to.

        Examples:
            Write some secret data.
            ```python
            secrets_manager = SecretsManager.load("MY_BLOCK")
            secrets_manager.write_secret(b"my_secret_data")
            ```
        """
        client = self.aws_credentials.get_secrets_manager_client()
        try:
            response = await run_sync_in_worker_thread(
                client.put_secret_value,
                SecretId=self.secret_name,
                SecretBinary=secret_data,
                **put_or_create_secret_kwargs,
            )
        except client.exceptions.ResourceNotFoundException:
            self.logger.info(
                f"The secret {self.secret_name!r} does not exist yet, creating it now."
            )
            response = await run_sync_in_worker_thread(
                client.create_secret,
                Name=self.secret_name,
                SecretBinary=secret_data,
                **put_or_create_secret_kwargs,
            )
        arn = response["ARN"]
        self.logger.info(f"The secret data was written successfully to {arn!r}.")
        return arn

    @sync_compatible
    async def delete_secret(
        self,
        recovery_window_in_days: int = 30,
        force_delete_without_recovery: bool = False,
        **delete_kwargs: Dict[str, Any],
    ) -> str:
        """
        Deletes the secret from the secret storage service.

        Args:
            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.

        Returns:
            The path that the secret was deleted from.

        Examples:
            Deletes the secret with a recovery window of 15 days.
            ```python
            secrets_manager = SecretsManager.load("MY_BLOCK")
            secrets_manager.delete_secret(recovery_window_in_days=15)
            ```
        """
        if force_delete_without_recovery and recovery_window_in_days:
            raise ValueError(
                "Cannot specify recovery window and force delete without recovery."
            )
        elif not (7 <= recovery_window_in_days <= 30):
            raise ValueError(
                "Recovery window must be between 7 and 30 days, got "
                f"{recovery_window_in_days}."
            )

        client = self.aws_credentials.get_secrets_manager_client()
        response = await run_sync_in_worker_thread(
            client.delete_secret,
            SecretId=self.secret_name,
            RecoveryWindowInDays=recovery_window_in_days,
            ForceDeleteWithoutRecovery=force_delete_without_recovery,
            **delete_kwargs,
        )
        arn = response["ARN"]
        self.logger.info(f"The secret {arn} was deleted successfully.")
        return arn

Attributes

secret_name: str pydantic-field required

The name of the secret.

Methods

__json_encoder__ special staticmethod

partial(func, args, *keywords) - new function with partial application of the given arguments and keywords.

delete_secret async

Deletes the secret from the secret storage service.

Parameters:

Name	Type	Description	Default
recovery_window_in_days	int	The number of days to wait before permanently deleting the secret. Must be between 7 and 30 days.	30
force_delete_without_recovery	bool	If True, the secret will be deleted immediately without a recovery window.	False
**delete_kwargs	Dict[str, Any]	Additional keyword arguments to pass to the delete_secret method of the boto3 client.	{}

Returns:

Type	Description
str	The path that the secret was deleted from.

Examples:

Deletes the secret with a recovery window of 15 days.

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

Source code in prefect_aws/secrets_manager.py

@sync_compatible
async def delete_secret(
    self,
    recovery_window_in_days: int = 30,
    force_delete_without_recovery: bool = False,
    **delete_kwargs: Dict[str, Any],
) -> str:
    """
    Deletes the secret from the secret storage service.

    Args:
        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.

    Returns:
        The path that the secret was deleted from.

    Examples:
        Deletes the secret with a recovery window of 15 days.
        ```python
        secrets_manager = SecretsManager.load("MY_BLOCK")
        secrets_manager.delete_secret(recovery_window_in_days=15)
        ```
    """
    if force_delete_without_recovery and recovery_window_in_days:
        raise ValueError(
            "Cannot specify recovery window and force delete without recovery."
        )
    elif not (7 <= recovery_window_in_days <= 30):
        raise ValueError(
            "Recovery window must be between 7 and 30 days, got "
            f"{recovery_window_in_days}."
        )

    client = self.aws_credentials.get_secrets_manager_client()
    response = await run_sync_in_worker_thread(
        client.delete_secret,
        SecretId=self.secret_name,
        RecoveryWindowInDays=recovery_window_in_days,
        ForceDeleteWithoutRecovery=force_delete_without_recovery,
        **delete_kwargs,
    )
    arn = response["ARN"]
    self.logger.info(f"The secret {arn} was deleted successfully.")
    return arn

read_secret async

Reads the secret from the secret storage service.

Parameters:

Name	Type	Description	Default
version_id	str	The version of the secret to read. If not provided, the latest version will be read.	None
version_stage	str	The version stage of the secret to read. If not provided, the latest version will be read.	None
read_kwargs	Dict[str, Any]	Additional keyword arguments to pass to the get_secret_value method of the boto3 client.	{}

Returns:

Type	Description
bytes	The secret data.

Examples:

Reads a secret.

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

Source code in prefect_aws/secrets_manager.py

@sync_compatible
async def read_secret(
    self,
    version_id: str = None,
    version_stage: str = None,
    **read_kwargs: Dict[str, Any],
) -> bytes:
    """
    Reads the secret from the secret storage service.

    Args:
        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.

    Returns:
        The secret data.

    Examples:
        Reads a secret.
        ```python
        secrets_manager = SecretsManager.load("MY_BLOCK")
        secrets_manager.read_secret()
        ```
    """
    client = self.aws_credentials.get_secrets_manager_client()
    if version_id is not None:
        read_kwargs["VersionId"] = version_id
    if version_stage is not None:
        read_kwargs["VersionStage"] = version_stage
    response = await run_sync_in_worker_thread(
        client.get_secret_value, SecretId=self.secret_name, **read_kwargs
    )
    if "SecretBinary" in response:
        secret = response["SecretBinary"]
    elif "SecretString" in response:
        secret = response["SecretString"]
    arn = response["ARN"]
    self.logger.info(f"The secret {arn!r} data was successfully read.")
    return secret

write_secret async

Writes the secret to the secret storage service as a SecretBinary; if it doesn't exist, it will be created.

Parameters:

Name	Type	Description	Default
secret_data	bytes	The secret data to write.	required
**put_or_create_secret_kwargs	Dict[str, Any]	Additional keyword arguments to pass to put_secret_value or create_secret method of the boto3 client.	{}

Returns:

Type	Description
str	The path that the secret was written to.

Examples:

Write some secret data.

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

Source code in prefect_aws/secrets_manager.py

@sync_compatible
async def write_secret(
    self, secret_data: bytes, **put_or_create_secret_kwargs: Dict[str, Any]
) -> str:
    """
    Writes the secret to the secret storage service as a SecretBinary;
    if it doesn't exist, it will be created.

    Args:
        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.

    Returns:
        The path that the secret was written to.

    Examples:
        Write some secret data.
        ```python
        secrets_manager = SecretsManager.load("MY_BLOCK")
        secrets_manager.write_secret(b"my_secret_data")
        ```
    """
    client = self.aws_credentials.get_secrets_manager_client()
    try:
        response = await run_sync_in_worker_thread(
            client.put_secret_value,
            SecretId=self.secret_name,
            SecretBinary=secret_data,
            **put_or_create_secret_kwargs,
        )
    except client.exceptions.ResourceNotFoundException:
        self.logger.info(
            f"The secret {self.secret_name!r} does not exist yet, creating it now."
        )
        response = await run_sync_in_worker_thread(
            client.create_secret,
            Name=self.secret_name,
            SecretBinary=secret_data,
            **put_or_create_secret_kwargs,
        )
    arn = response["ARN"]
    self.logger.info(f"The secret data was written successfully to {arn!r}.")
    return arn

Functions

create_secret async

Creates a secret in AWS Secrets Manager.

Parameters:

Name	Type	Description	Default
secret_name	str	The name of the secret to create.	required
secret_value	Union[str, bytes]	The value to store in the created secret.	required
aws_credentials	AwsCredentials	Credentials to use for authentication with AWS.	required
description	Optional[str]	A description for the created secret.	None
tags	Optional[List[Dict[str, str]]]	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 }	None

Returns:

Type	Description
A dict containing the secret ARN (Amazon Resource Name), name, and current version ID. ```python { "ARN"	str, "Name": str, "VersionId": str } ```

Examples:

Create a secret:

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()

Source code in prefect_aws/secrets_manager.py

@task
async def 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]:
    """
    Creates a secret in AWS Secrets Manager.

    Args:
        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:
            ```python
            {
                "Key": str,
                "Value": str
            }
            ```

    Returns:
        A dict containing the secret ARN (Amazon Resource Name),
            name, and current version ID.
            ```python
            {
                "ARN": str,
                "Name": str,
                "VersionId": str
            }
            ```
    Example:
        Create a secret:

        ```python
        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()
        ```


    """
    create_secret_kwargs: Dict[str, Union[str, bytes, List[Dict[str, str]]]] = dict(
        Name=secret_name
    )
    if description is not None:
        create_secret_kwargs["Description"] = description
    if tags is not None:
        create_secret_kwargs["Tags"] = tags
    if isinstance(secret_value, bytes):
        create_secret_kwargs["SecretBinary"] = secret_value
    elif isinstance(secret_value, str):
        create_secret_kwargs["SecretString"] = secret_value
    else:
        raise ValueError("Please provide a bytes or str value for secret_value")

    logger = get_run_logger()
    logger.info("Creating secret named %s", secret_name)

    client = aws_credentials.get_boto3_session().client("secretsmanager")

    try:
        response = await run_sync_in_worker_thread(
            client.create_secret, **create_secret_kwargs
        )
        print(response.pop("ResponseMetadata", None))
        return response
    except ClientError:
        logger.exception("Unable to create secret %s", secret_name)
        raise

delete_secret async

Deletes a secret from AWS Secrets Manager.

Secrets can either be deleted immediately by setting force_delete_without_recovery equal to True. Otherwise, secrets will be marked for deletion and available for recovery for the number of days specified in recovery_window_in_days

Parameters:

Name	Type	Description	Default
secret_name	str	Name of the secret to be deleted.	required
aws_credentials	AwsCredentials	Credentials to use for authentication with AWS.	required
recovery_window_in_days	int	Number of days a secret should be recoverable for before permanent deletion. Minium window is 7 days and maximum window is 30 days. If force_delete_without_recovery is set to True, this value will be ignored.	30
force_delete_without_recovery	bool	If True, the secret will be immediately deleted and will not be recoverable.	False

Returns:

Type	Description
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`. ```python { "ARN"	str, "Name": str, "DeletionDate": datetime.datetime } ```

Examples:

Delete a secret immediately:

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()

Delete a secret with a 90 day recovery window:

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()

Source code in prefect_aws/secrets_manager.py

@task
async def delete_secret(
    secret_name: str,
    aws_credentials: AwsCredentials,
    recovery_window_in_days: int = 30,
    force_delete_without_recovery: bool = False,
) -> Dict[str, str]:
    """
    Deletes a secret from AWS Secrets Manager.

    Secrets can either be deleted immediately by setting `force_delete_without_recovery`
    equal to `True`. Otherwise, secrets will be marked for deletion and available for
    recovery for the number of days specified in `recovery_window_in_days`

    Args:
        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. Minium 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.

    Returns:
        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`.
            ```python
            {
                "ARN": str,
                "Name": str,
                "DeletionDate": datetime.datetime
            }
            ```

    Examples:
        Delete a secret immediately:

        ```python
        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()
        ```

        Delete a secret with a 90 day recovery window:

        ```python
        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()
        ```


    """
    if not force_delete_without_recovery and not (7 <= recovery_window_in_days <= 30):
        raise ValueError("Recovery window must be between 7 and 30 days.")

    delete_secret_kwargs: Dict[str, Union[str, int, bool]] = dict(SecretId=secret_name)
    if force_delete_without_recovery:
        delete_secret_kwargs["ForceDeleteWithoutRecovery"] = (
            force_delete_without_recovery
        )
    else:
        delete_secret_kwargs["RecoveryWindowInDays"] = recovery_window_in_days

    logger = get_run_logger()
    logger.info("Deleting secret %s", secret_name)

    client = aws_credentials.get_boto3_session().client("secretsmanager")

    try:
        response = await run_sync_in_worker_thread(
            client.delete_secret, **delete_secret_kwargs
        )
        response.pop("ResponseMetadata", None)
        return response
    except ClientError:
        logger.exception("Unable to delete secret %s", secret_name)
        raise

read_secret async

Reads the value of a given secret from AWS Secrets Manager.

Parameters:

Name	Type	Description	Default
secret_name	str	Name of stored secret.	required
aws_credentials	AwsCredentials	Credentials to use for authentication with AWS.	required
version_id	Optional[str]	Specifies version of secret to read. Defaults to the most recent version if not given.	None
version_stage	Optional[str]	Specifies the version stage of the secret to read. Defaults to AWS_CURRENT if not given.	None

Returns:

Type	Description
Union[str, bytes]	The secret values as a str or bytes depending on the format in which the secret was stored.

Examples:

Read a secret value:

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

@flow
def example_read_secret():
    aws_credentials = AwsCredentials(
        aws_access_key_id="access_key_id",
        aws_secret_access_key="secret_access_key"
    )
    secret_value = read_secret(
        secret_name="db_password",
        aws_credentials=aws_credentials
    )

example_read_secret()

Source code in prefect_aws/secrets_manager.py

@task
async def read_secret(
    secret_name: str,
    aws_credentials: AwsCredentials,
    version_id: Optional[str] = None,
    version_stage: Optional[str] = None,
) -> Union[str, bytes]:
    """
    Reads the value of a given secret from AWS Secrets Manager.

    Args:
        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.

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

    Example:
        Read a secret value:

        ```python
        from prefect import flow
        from prefect_aws import AwsCredentials
        from prefect_aws.secrets_manager import read_secret

        @flow
        def example_read_secret():
            aws_credentials = AwsCredentials(
                aws_access_key_id="access_key_id",
                aws_secret_access_key="secret_access_key"
            )
            secret_value = read_secret(
                secret_name="db_password",
                aws_credentials=aws_credentials
            )

        example_read_secret()
        ```
    """
    logger = get_run_logger()
    logger.info("Getting value for secret %s", secret_name)

    client = aws_credentials.get_boto3_session().client("secretsmanager")

    get_secret_value_kwargs = dict(SecretId=secret_name)
    if version_id is not None:
        get_secret_value_kwargs["VersionId"] = version_id
    if version_stage is not None:
        get_secret_value_kwargs["VersionStage"] = version_stage

    try:
        response = await run_sync_in_worker_thread(
            client.get_secret_value, **get_secret_value_kwargs
        )
    except ClientError:
        logger.exception("Unable to get value for secret %s", secret_name)
        raise
    else:
        return response.get("SecretString") or response.get("SecretBinary")

update_secret async

Updates the value of a given secret in AWS Secrets Manager.

Parameters:

Name	Type	Description	Default
secret_name	str	Name of secret to update.	required
secret_value	Union[str, bytes]	Desired value of the secret. Can be either str or bytes.	required
aws_credentials	AwsCredentials	Credentials to use for authentication with AWS.	required
description	Optional[str]	Desired description of the secret.	None

Returns:

Type	Description
A dict containing the secret ARN (Amazon Resource Name), name, and current version ID. ```python { "ARN"	str, "Name": str, "VersionId": str } ```

Examples:

Update a secret value:

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

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

example_update_secret()

Source code in prefect_aws/secrets_manager.py

@task
async def update_secret(
    secret_name: str,
    secret_value: Union[str, bytes],
    aws_credentials: AwsCredentials,
    description: Optional[str] = None,
) -> Dict[str, str]:
    """
    Updates the value of a given secret in AWS Secrets Manager.

    Args:
        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.

    Returns:
        A dict containing the secret ARN (Amazon Resource Name),
            name, and current version ID.
            ```python
            {
                "ARN": str,
                "Name": str,
                "VersionId": str
            }
            ```

    Example:
        Update a secret value:

        ```python
        from prefect import flow
        from prefect_aws import AwsCredentials
        from prefect_aws.secrets_manager import update_secret

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

        example_update_secret()
        ```

    """
    update_secret_kwargs: Dict[str, Union[str, bytes]] = dict(SecretId=secret_name)
    if description is not None:
        update_secret_kwargs["Description"] = description
    if isinstance(secret_value, bytes):
        update_secret_kwargs["SecretBinary"] = secret_value
    elif isinstance(secret_value, str):
        update_secret_kwargs["SecretString"] = secret_value
    else:
        raise ValueError("Please provide a bytes or str value for secret_value")

    logger = get_run_logger()
    logger.info("Updating value for secret %s", secret_name)

    client = aws_credentials.get_boto3_session().client("secretsmanager")

    try:
        response = await run_sync_in_worker_thread(
            client.update_secret, **update_secret_kwargs
        )
        response.pop("ResponseMetadata", None)
        return response
    except ClientError:
        logger.exception("Unable to update secret %s", secret_name)
        raise