rds

Abstract dataclass for RDS DB instance, cluster.

class simple_aws_rds.rds.RDSDBInstanceStatusEnum(value)

RDS DB instance status enum. The official document doesn’t mention the exact string. But I found it is all lower case slugified.

Reference:

• https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/accessing-monitoring.html#Overview.DBInstance.Status

class simple_aws_rds.rds.RDSDBInstanceStatusGroupEnum

Group RDS DB instance status enum by semantic.

Parameters:

• ended – The status that the instance can be considered as “stopped”, “won’t change”.

• in_transition – The status that the instance can be considered as “changing”.

• impossible_to_become_available – as the name

• impossible_to_become_stopped – as the name

class simple_aws_rds.rds.RDSDBInstance(id: str, status: ~typing.Optional[str] = None, instance_class: ~typing.Optional[str] = None, instance_create_time: ~typing.Optional[~datetime.datetime] = None, engine: ~typing.Optional[str] = None, engine_version: ~typing.Optional[str] = None, endpoint: ~typing.Optional[str] = None, port: ~typing.Optional[int] = None, hosted_zone_id: ~typing.Optional[str] = None, master_username: ~typing.Optional[str] = None, allocated_storage: ~typing.Optional[str] = None, preferred_backup_window: ~typing.Optional[str] = None, backup_retention_period: ~typing.Optional[int] = None, preferred_maintenance_window: ~typing.Optional[str] = None, latest_restorable_time: ~typing.Optional[~datetime.datetime] = None, multi_az: ~typing.Optional[bool] = None, auto_minor_version_upgrade: ~typing.Optional[bool] = None, replica_mode: ~typing.Optional[str] = None, license_model: ~typing.Optional[str] = None, iops: ~typing.Optional[int] = None, character_set_name: ~typing.Optional[str] = None, secondary_availability_zone: ~typing.Optional[str] = None, publicly_accessible: ~typing.Optional[bool] = None, storage_type: ~typing.Optional[str] = None, tde_credential_arn: ~typing.Optional[str] = None, db_instance_port: ~typing.Optional[int] = None, db_cluster_identifier: ~typing.Optional[str] = None, storage_encrypted: ~typing.Optional[bool] = None, kms_key_id: ~typing.Optional[str] = None, dbi_resource_id: ~typing.Optional[str] = None, ca_certificate_identifier: ~typing.Optional[str] = None, copy_tags_to_snapshot: ~typing.Optional[bool] = None, monitoring_interval: ~typing.Optional[int] = None, enhanced_monitoring_resource_arn: ~typing.Optional[str] = None, monitoring_role_arn: ~typing.Optional[str] = None, promotion_tier: ~typing.Optional[int] = None, db_instance_arn: ~typing.Optional[str] = None, timezone: ~typing.Optional[str] = None, iam_database_authentication_enabled: ~typing.Optional[bool] = None, performance_insights_enabled: ~typing.Optional[bool] = None, performance_insights_kms_key_id: ~typing.Optional[str] = None, performance_insights_retention_period: ~typing.Optional[int] = None, deletion_protection: ~typing.Optional[bool] = None, max_allocated_storage: ~typing.Optional[int] = None, customer_owned_ip_enabled: ~typing.Optional[bool] = None, aws_backup_recovery_point_arn: ~typing.Optional[str] = None, activity_stream_status: ~typing.Optional[str] = None, activity_stream_kms_key_id: ~typing.Optional[str] = None, activity_stream_kinesis_stream_name: ~typing.Optional[str] = None, activity_stream_mode: ~typing.Optional[str] = None, activity_stream_engine_native_audit_fields_included: ~typing.Optional[bool] = None, automation_mode: ~typing.Optional[str] = None, resume_full_automation_mode_time: ~typing.Optional[~datetime.datetime] = None, custom_iam_instance_profile: ~typing.Optional[str] = None, backup_target: ~typing.Optional[str] = None, network_type: ~typing.Optional[str] = None, activity_stream_policy_status: ~typing.Optional[str] = None, storage_throughput: ~typing.Optional[int] = None, db_system_id: ~typing.Optional[str] = None, read_replica_source_db_cluster_identifier: ~typing.Optional[str] = None, vpc_id: ~typing.Optional[str] = None, subnet_ids: ~typing.List[str] = <factory>, subnet_group_name: ~typing.Optional[str] = None, subnet_group_description: ~typing.Optional[str] = None, subnet_group_arn: ~typing.Optional[str] = None, subnet_group_status: ~typing.Optional[str] = None, security_groups: ~typing.List[~typing.Dict[str, str]] = <factory>, availability_zone: ~typing.Optional[str] = None, tags: ~typing.Dict[str, str] = <factory>, data: ~typing.Dict[str, ~typing.Any] = <factory>)

Represent an RDS DB instance.

See attributes explanation at: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/docdb/client/describe_db_instances.html

classmethod from_dict(dct: dict)

Create an RDS DB instance object from the describe_db_instances API response.

start_db_instance(rds_client: RDSClient) → StartDBInstanceResultTypeDef

Start the RDS DB instance

stop_db_instance(rds_client: RDSClient) → StopDBInstanceResultTypeDef

Stop the RDS DB instance

delete_db_instance(rds_client: RDSClient, skip_final_snapshot: bool = Sentinel('NOTHING'), final_db_snapshot_identifier: str = Sentinel('NOTHING'), delete_automated_backups: bool = Sentinel('NOTHING')) → DeleteDBInstanceResultTypeDef

Delete the RDS DB instance

wait_for_status(rds_client: RDSClient, stop_status: Union[RDSDBInstanceStatusEnum, List[RDSDBInstanceStatusEnum]], gap: Union[int, float] = 1, delays: Union[int, float] = 10, timeout: Union[int, float] = 300, instant: bool = True, error_status: Optional[Union[RDSDBInstanceStatusEnum, List[RDSDBInstanceStatusEnum]]] = None, indent: int = 0, verbose: bool = True) → RDSDBInstance

wait until the DB instance reaches the specified status defined in stop_status. If reaches any of ``error_status ``, raise error.

Parameters:

• rds_client –

• stop_status – status to stop waiting

• gap – the time to wait before making first status check

• delays – delay between each check

• timeout – timeout in seconds

• instant – if True, then the first check is instant

• error_status – status to raise error

• indent – indent level for logging

• verbose – whether to print log

Returns:

the RDSDBInstance representing the latest status of DB instance.

wait_for_available(rds_client: RDSClient, gap: Union[int, float] = 1, delays: Union[int, float] = 10, timeout: Union[int, float] = 300, instant: bool = True, indent: int = 0, verbose: bool = True) → RDSDBInstance

Similar to RDSDBInstance.wait_for_status(), but wait for DB instance to reach “available” status.

wait_for_stopped(rds_client: RDSClient, gap: Union[int, float] = 1, delays: Union[int, float] = 10, timeout: Union[int, float] = 300, instant: bool = True, indent: int = 0, verbose: bool = True) → RDSDBInstance

Similar to RDSDBInstance.wait_for_status(), but wait for DB instance to reach “stopped” status.

classmethod query(rds_client: RDSClient, db_instance_identifier: str = Sentinel('NOTHING'), filters: List[dict] = Sentinel('NOTHING')) → RDSDBInstanceIterProxy

A wrapper around rds_client.describe_db_instances.

Multiple filters join with logic “AND”, multiple values in a filter join with logic “OR”.

classmethod from_id(rds_client: RDSClient, db_identifier: str) → Optional[RDSDBInstance]

Get RDS DB instance details by it’s id.

classmethod from_tag_key_value(rds_client: RDSClient, key: str, value: str) → RDSDBInstanceIterProxy

Query RDS DB Instance by tag key and value. This function only support single key, value pair filtering, if you want more advanced filtering, you can use the query method and do in-memory filtering.

Parameters:

• key – tag key

• value – tag value

class simple_aws_rds.rds.RDSDBInstanceIterProxy(iterable: Iterable)

IterProxy for RDSDBInstance.

class simple_aws_rds.rds.RDSDBSnapshotStatusEnum(value)

RDS DB snapshot status enum. The official document doesn’t mention the exact string. But I found it is all lower case slugified.

Reference:

• https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER_CreateSnapshot.html

class simple_aws_rds.rds.RDSDBSnapshot(db_snapshot_identifier: str, db_instance_identifier: ~typing.Optional[str] = None, snapshot_create_time: ~typing.Optional[~datetime.datetime] = None, engine: ~typing.Optional[str] = None, allocated_storage: ~typing.Optional[int] = None, status: ~typing.Optional[str] = None, port: ~typing.Optional[int] = None, availability_zone: ~typing.Optional[str] = None, vpc_id: ~typing.Optional[str] = None, instance_create_time: ~typing.Optional[~datetime.datetime] = None, master_username: ~typing.Optional[str] = None, engine_version: ~typing.Optional[str] = None, license_model: ~typing.Optional[str] = None, snapshot_type: ~typing.Optional[str] = None, iops: ~typing.Optional[str] = None, option_group_name: ~typing.Optional[str] = None, percent_progress: ~typing.Optional[int] = None, source_region: ~typing.Optional[str] = None, source_db_snapshot_identifier: ~typing.Optional[str] = None, storage_type: ~typing.Optional[str] = None, tde_credential_arn: ~typing.Optional[str] = None, encrypted: ~typing.Optional[bool] = None, kms_key_id: ~typing.Optional[str] = None, db_snapshot_arn: ~typing.Optional[str] = None, timezone: ~typing.Optional[str] = None, iam_database_authentication_enabled: ~typing.Optional[bool] = None, processor_features: ~typing.List[~typing.Dict[str, str]] = <factory>, dbi_resource_id: ~typing.Optional[str] = None, tags: ~typing.Dict[str, str] = <factory>, original_snapshot_create_time: ~typing.Optional[~datetime.datetime] = None, snapshot_database_time: ~typing.Optional[~datetime.datetime] = None, snapshot_target: ~typing.Optional[str] = None, storage_throughput: ~typing.Optional[str] = None, db_system_id: ~typing.Optional[str] = None, dedicated_log_volume: ~typing.Optional[str] = None, multi_tenant: ~typing.Optional[str] = None, data: ~typing.Dict[str, ~typing.Any] = <factory>)

Represent an RDS DB instance.

See attributes explanation at: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/rds/client/describe_db_snapshots.html

is_creating() → bool

Check if the snapshot is in creating status.

is_available() → bool

Check if the snapshot is in available status.

wait_for_status(rds_client: RDSClient, stop_status: Union[RDSDBSnapshotStatusEnum, List[RDSDBSnapshotStatusEnum]], gap: Union[int, float] = 1, delays: Union[int, float] = 10, timeout: Union[int, float] = 300, instant: bool = True, error_status: Optional[Union[RDSDBSnapshotStatusEnum, List[RDSDBSnapshotStatusEnum]]] = None, indent: int = 0, verbose: bool = True) → RDSDBSnapshot

wait until the DB snapshot reaches the specified status defined in stop_status. If reaches any of ``error_status ``, raise error.

Parameters:

• rds_client –

• stop_status – status to stop waiting

• gap – the time to wait before making first status check

• delays – delay between each check

• timeout – timeout in seconds

• instant – if True, then the first check is instant

• error_status – status to raise error

• indent – indent level for logging

• verbose – whether to print log

Returns:

the RDSDBInstance representing the latest status of DB instance.

wait_for_available(rds_client, gap: Union[int, float] = 1, delays: Union[int, float] = 10, timeout: Union[int, float] = 300, instant: bool = True, indent: int = 0, verbose: bool = True) → RDSDBSnapshot

Similar to RDSDBSnapshot.wait_for_status(), but wait for DB snapshot to reach “available” status.

delete_db_snapshot(rds_client: RDSClient) → DeleteDBSnapshotResultTypeDef

Delete the RDS DB instance

classmethod query(rds_client: RDSClient, db_instance_identifier: str = Sentinel('NOTHING'), db_snapshot_identifier: str = Sentinel('NOTHING'), filters: List[dict] = Sentinel('NOTHING'), snapshot_type: str = Sentinel('NOTHING'), include_shared: bool = Sentinel('NOTHING'), include_public: bool = Sentinel('NOTHING'), dbi_resource_id: str = Sentinel('NOTHING')) → RDSDBSnapshotIterProxy

A wrapper around rds_client.describe_db_snapshots.

Multiple filters join with logic “AND”, multiple values in a filter join with logic “OR”.

classmethod from_id(rds_client: RDSClient, db_snapshot_identifier: str) → Optional[RDSDBInstance]

Get RDS DB snapshot details by it’s id.

classmethod from_tag_key_value(rds_client: RDSClient, key: str, value: str) → RDSDBSnapshotIterProxy

Query RDS DB snapshot by tag key and value. This function only support single key, value pair filtering, if you want more advanced filtering, you can use the query method and do in-memory filtering.

Parameters:

• key – tag key

• value – tag value

class simple_aws_rds.rds.RDSDBSnapshotIterProxy(iterable: Iterable)

IterProxy for RDSDBSnapshot.