rboto-secrets-manager

rboto is an independent project maintained by Ferrumio. It is not affiliated with, sponsored by, or endorsed by Amazon Web Services (AWS).

Preview (Alpha). rboto is under active development. APIs may change between releases.

rboto provides an async-only, typed Python client for AWS Secrets Manager, powered by Rust. This package contains SecretsManagerClient and 23 operations.

Installation

Requires Python 3.12 or newer and a native wheel for your Python version and platform.

python -m pip install "rboto[secrets-manager]"

You can also install rboto-secrets-manager directly; it installs the matching rboto package. Only the selected service is installed.

These are the package names to use when a release is available on your package index. If your index has no release yet, follow the source installation instructions.

Usage

Configure credentials through your environment, shared profile, or workload role. The example uses us-east-1; select a region where your service and resources are available. Resource names, identifiers, input files, and model IDs in the example are placeholders to replace with your own.

import asyncio
import json

from rboto import SecretsManagerClient


async def main() -> None:
    client = SecretsManagerClient(region="us-east-1")
    response = await client.get_secret_value(secret_id="application/config")
    if response.secret_string is None:
        raise ValueError("Expected a text secret")
    # Pass the value to your application; do not print secret contents.
    settings = json.loads(response.secret_string)
    print("Loaded settings:", len(settings))


if __name__ == "__main__":
    asyncio.run(main())

In an application with an event loop, call await main() instead of asyncio.run(main()). Reuse the client across calls. Constructing a client does not send a request.

API conventions

• Import the client with from rboto import SecretsManagerClient.

• Every operation is asynchronous: use await client.method(...).

• Operation arguments are keyword-only and use the Python names listed below.

• Structured inputs use typed dictionaries and tagged unions from rboto_secrets_manager.types.

• Regular responses expose typed attributes; response.to_dict() converts them to a dictionary.

• Event-stream responses may be typed dictionaries containing async iterators. Follow the declared return type; consume event streams with async for.

• Byte streams are consumed asynchronously. await stream.read() buffers the complete body; iterate the stream when processing a large response.

• Direct operation calls return one page.

• Use async for page in client.paginators.method(...) for supported paginated operations. Paginators accept the same typed arguments as the corresponding client method.

• Catch modeled exceptions from rboto_secrets_manager.exceptions. Their error_code, message, and operation_name identify service failures.

Clients also accept profile, endpoint_url, or shared Session settings. See configuration, errors, and streaming.

Operations

This is the complete operation catalog for this generated package. “Required arguments” lists arguments without defaults in the Python signature. Service-side conditions may require additional arguments. Each method links to its complete signature, including optional arguments. Nested input definitions are in rboto_secrets_manager.types; native response definitions are in the package's _native.pyi.

Method	Required arguments	Return type
batch_get_secret_value	None	BatchGetSecretValueOutput
cancel_rotate_secret	secret_id: str	CancelRotateSecretOutput
create_secret	name: str	CreateSecretOutput
delete_resource_policy	secret_id: str	DeleteResourcePolicyOutput
delete_secret	secret_id: str	DeleteSecretOutput
describe_secret	secret_id: str	DescribeSecretOutput
get_random_password	None	GetRandomPasswordOutput
get_resource_policy	secret_id: str	GetResourcePolicyOutput
get_secret_value	secret_id: str	GetSecretValueOutput
list_secret_version_ids	secret_id: str	ListSecretVersionIdsOutput
list_secrets	None	ListSecretsOutput
put_resource_policy	secret_id: str, resource_policy: str	PutResourcePolicyOutput
put_secret_value	secret_id: str	PutSecretValueOutput
remove_regions_from_replication	secret_id: str, remove_replica_regions: Sequence[str]	RemoveRegionsFromReplicationOutput
replicate_secret_to_regions	secret_id: str, add_replica_regions: Sequence[ReplicaRegionType]	ReplicateSecretToRegionsOutput
restore_secret	secret_id: str	RestoreSecretOutput
rotate_secret	secret_id: str	RotateSecretOutput
stop_replication_to_replica	secret_id: str	StopReplicationToReplicaOutput
tag_resource	secret_id: str, tags: Sequence[Tag]	None
untag_resource	secret_id: str, tag_keys: Sequence[str]	None
update_secret	secret_id: str	UpdateSecretOutput
update_secret_version_stage	secret_id: str, version_stage: str	UpdateSecretVersionStageOutput
validate_resource_policy	resource_policy: str	ValidateResourcePolicyOutput

Complete signatures

batch_get_secret_value

async def batch_get_secret_value(
    *,
    secret_id_list: Sequence[str] | None = None,
    filters: Sequence[Filter] | None = None,
    max_results: int | None = None,
    next_token: str | None = None,
) -> BatchGetSecretValueOutput: ...

Also available as client.paginators.batch_get_secret_value(...), returning an async iterator of the response type above.

cancel_rotate_secret

async def cancel_rotate_secret(
    *,
    secret_id: str,
) -> CancelRotateSecretOutput: ...

create_secret

async def create_secret(
    *,
    name: str,
    client_request_token: str | None = None,
    description: str | None = None,
    kms_key_id: str | None = None,
    secret_binary: bytes | None = None,
    secret_string: str | None = None,
    tags: Sequence[Tag] | None = None,
    add_replica_regions: Sequence[ReplicaRegionType] | None = None,
    force_overwrite_replica_secret: bool | None = None,
    type: str | None = None,
) -> CreateSecretOutput: ...

delete_resource_policy

async def delete_resource_policy(
    *,
    secret_id: str,
) -> DeleteResourcePolicyOutput: ...

delete_secret

async def delete_secret(
    *,
    secret_id: str,
    recovery_window_in_days: int | None = None,
    force_delete_without_recovery: bool | None = None,
) -> DeleteSecretOutput: ...

describe_secret

async def describe_secret(
    *,
    secret_id: str,
) -> DescribeSecretOutput: ...

get_random_password

async def get_random_password(
    *,
    password_length: int | None = None,
    exclude_characters: str | None = None,
    exclude_numbers: bool | None = None,
    exclude_punctuation: bool | None = None,
    exclude_uppercase: bool | None = None,
    exclude_lowercase: bool | None = None,
    include_space: bool | None = None,
    require_each_included_type: bool | None = None,
) -> GetRandomPasswordOutput: ...

get_resource_policy

async def get_resource_policy(
    *,
    secret_id: str,
) -> GetResourcePolicyOutput: ...

get_secret_value

async def get_secret_value(
    *,
    secret_id: str,
    version_id: str | None = None,
    version_stage: str | None = None,
) -> GetSecretValueOutput: ...

list_secret_version_ids

async def list_secret_version_ids(
    *,
    secret_id: str,
    max_results: int | None = None,
    next_token: str | None = None,
    include_deprecated: bool | None = None,
) -> ListSecretVersionIdsOutput: ...

Also available as client.paginators.list_secret_version_ids(...), returning an async iterator of the response type above.

list_secrets

async def list_secrets(
    *,
    include_planned_deletion: bool | None = None,
    max_results: int | None = None,
    next_token: str | None = None,
    filters: Sequence[Filter] | None = None,
    sort_order: SortOrderType | None = None,
    sort_by: SortByType | None = None,
) -> ListSecretsOutput: ...

Also available as client.paginators.list_secrets(...), returning an async iterator of the response type above.

put_resource_policy

async def put_resource_policy(
    *,
    secret_id: str,
    resource_policy: str,
    block_public_policy: bool | None = None,
) -> PutResourcePolicyOutput: ...

put_secret_value

async def put_secret_value(
    *,
    secret_id: str,
    client_request_token: str | None = None,
    secret_binary: bytes | None = None,
    secret_string: str | None = None,
    version_stages: Sequence[str] | None = None,
    rotation_token: str | None = None,
) -> PutSecretValueOutput: ...

remove_regions_from_replication

async def remove_regions_from_replication(
    *,
    secret_id: str,
    remove_replica_regions: Sequence[str],
) -> RemoveRegionsFromReplicationOutput: ...

replicate_secret_to_regions

async def replicate_secret_to_regions(
    *,
    secret_id: str,
    add_replica_regions: Sequence[ReplicaRegionType],
    force_overwrite_replica_secret: bool | None = None,
) -> ReplicateSecretToRegionsOutput: ...

restore_secret

async def restore_secret(
    *,
    secret_id: str,
) -> RestoreSecretOutput: ...

rotate_secret

async def rotate_secret(
    *,
    secret_id: str,
    client_request_token: str | None = None,
    rotation_lambda_arn: str | None = None,
    rotation_rules: RotationRulesType | None = None,
    external_secret_rotation_metadata: Sequence[ExternalSecretRotationMetadataItem] | None = None,
    external_secret_rotation_role_arn: str | None = None,
    rotate_immediately: bool | None = None,
) -> RotateSecretOutput: ...

stop_replication_to_replica

async def stop_replication_to_replica(
    *,
    secret_id: str,
) -> StopReplicationToReplicaOutput: ...

tag_resource

async def tag_resource(
    *,
    secret_id: str,
    tags: Sequence[Tag],
) -> None: ...

untag_resource

async def untag_resource(
    *,
    secret_id: str,
    tag_keys: Sequence[str],
) -> None: ...

update_secret

async def update_secret(
    *,
    secret_id: str,
    client_request_token: str | None = None,
    description: str | None = None,
    kms_key_id: str | None = None,
    secret_binary: bytes | None = None,
    secret_string: str | None = None,
    type: str | None = None,
) -> UpdateSecretOutput: ...

update_secret_version_stage

async def update_secret_version_stage(
    *,
    secret_id: str,
    version_stage: str,
    remove_from_version_id: str | None = None,
    move_to_version_id: str | None = None,
) -> UpdateSecretVersionStageOutput: ...

validate_resource_policy

async def validate_resource_policy(
    *,
    resource_policy: str,
    secret_id: str | None = None,
) -> ValidateResourcePolicyOutput: ...

Links

• Service guide
• Repository
• Issues
• Releases

Licensed under Apache-2.0.