# This file is automatically generated by botocraft. Do not edit directly.
# mypy: disable-error-code="index, override, assignment, union-attr, misc"
from .abstract import (
Boto3Model,
ReadonlyBoto3Model,
PrimaryBoto3Model,
ReadonlyPrimaryBoto3Model,
Boto3ModelManager,
ReadonlyBoto3ModelManager,
)
from botocraft.mixins.docdb import single_docdb_cluster_include_tags
from botocraft.services.kms import KMSKey, KMSKeyManager
from .abstract import PrimaryBoto3ModelQuerySet
from botocraft.mixins.docdb import single_docdb_subnet_group_include_tags
from datetime import datetime
from botocraft.mixins.docdb import multiple_docdb_cluster_include_tags
import builtins
from botocraft.mixins.docdb import DocDBClusterModelMixin
from botocraft.services.common import Filter
from botocraft.services.ec2 import SecurityGroup, SecurityGroupManager
from botocraft.mixins.docdb import multiple_docdb_instance_include_tags
from functools import cached_property
from pydantic import Field
from botocraft.mixins.docdb import multiple_docdb_subnet_group_include_tags
from collections import OrderedDict
from typing import ClassVar, Literal, Any, cast
from botocraft.services.ec2 import Vpc, VpcManager
from botocraft.mixins.tags import TagsDictMixin
from botocraft.services.ec2 import Subnet
from botocraft.mixins.docdb import single_docdb_instance_include_tags
from botocraft.services.common import Tag
# ===============
# Managers
# ===============
[docs]class DocDBClusterManager(Boto3ModelManager):
service_name: str = "docdb"
[docs] @single_docdb_cluster_include_tags
def create(
self,
model: "DocDBCluster",
DBClusterParameterGroupName: "str | None" = None,
VpcSecurityGroupIds: "builtins.list[str] | None" = None,
DBSubnetGroupName: "str | None" = None,
MasterUserPassword: "str | None" = None,
PreSignedUrl: "str | None" = None,
GlobalClusterIdentifier: "str | None" = None,
ManageMasterUserPassword: "bool | None" = None,
MasterUserSecretKmsKeyId: "str | None" = None,
SourceRegion: "str | None" = None,
) -> "DocDBCluster":
"""
Creates a new Amazon DocumentDB cluster.
Args:
model: The :py:class:`DBCluster` to create.
Keyword Args:
DBClusterParameterGroupName: The name of the cluster parameter group to associate with this cluster.
VpcSecurityGroupIds: A list of EC2 VPC security groups to associate with this cluster.
DBSubnetGroupName: A subnet group to associate with this cluster.
MasterUserPassword: The password for the master database user. This password can contain any printable ASCII
character except forward slash (/), double quote ("), or the "at" symbol (@).
PreSignedUrl: Not currently supported.
GlobalClusterIdentifier: The cluster identifier of the new global cluster.
ManageMasterUserPassword: Specifies whether to manage the master user password with Amazon Web Services Secrets
Manager.
MasterUserSecretKmsKeyId: The Amazon Web Services KMS key identifier to encrypt a secret that is automatically
generated and managed in Amazon Web Services Secrets Manager. This setting is valid only if the master user password
is managed by Amazon DocumentDB in Amazon Web Services Secrets Manager for the DB cluster.
SourceRegion: The ID of the region that contains the source for the db cluster.
"""
data = model.model_dump(exclude_none=True, by_alias=True)
args = dict(
DBClusterIdentifier=data.get("DBClusterIdentifier"),
Engine=data.get("Engine"),
AvailabilityZones=data.get("AvailabilityZones"),
BackupRetentionPeriod=data.get("BackupRetentionPeriod"),
DBClusterParameterGroupName=self.serialize(DBClusterParameterGroupName),
VpcSecurityGroupIds=self.serialize(VpcSecurityGroupIds),
DBSubnetGroupName=self.serialize(DBSubnetGroupName),
EngineVersion=data.get("EngineVersion"),
Port=data.get("Port"),
MasterUsername=data.get("MasterUsername"),
MasterUserPassword=self.serialize(MasterUserPassword),
PreferredBackupWindow=data.get("PreferredBackupWindow"),
PreferredMaintenanceWindow=data.get("PreferredMaintenanceWindow"),
Tags=data.get("Tags"),
StorageEncrypted=data.get("StorageEncrypted"),
KmsKeyId=data.get("KmsKeyId"),
PreSignedUrl=self.serialize(PreSignedUrl),
EnableCloudwatchLogsExports=data.get("EnabledCloudwatchLogsExports"),
DeletionProtection=data.get("DeletionProtection"),
GlobalClusterIdentifier=self.serialize(GlobalClusterIdentifier),
StorageType=data.get("StorageType"),
ServerlessV2ScalingConfiguration=data.get(
"ServerlessV2ScalingConfiguration"
),
ManageMasterUserPassword=self.serialize(ManageMasterUserPassword),
MasterUserSecretKmsKeyId=self.serialize(MasterUserSecretKmsKeyId),
NetworkType=data.get("NetworkType"),
SourceRegion=self.serialize(SourceRegion),
)
_response = self.client.create_db_cluster(
**{k: v for k, v in args.items() if v is not None}
)
response = CreateDBClusterResult(**_response)
self.sessionize(response.DBCluster)
return cast("DocDBCluster", response.DBCluster)
[docs] @single_docdb_cluster_include_tags
def update(
self,
model: "DocDBCluster",
NewDBClusterIdentifier: "str | None" = None,
ApplyImmediately: "bool | None" = None,
DBClusterParameterGroupName: "str | None" = None,
VpcSecurityGroupIds: "builtins.list[str] | None" = None,
MasterUserPassword: "str | None" = None,
CloudwatchLogsExportConfiguration: "CloudwatchLogsExportConfiguration | None" = None,
AllowMajorVersionUpgrade: "bool | None" = None,
ManageMasterUserPassword: "bool | None" = None,
MasterUserSecretKmsKeyId: "str | None" = None,
RotateMasterUserPassword: "bool | None" = None,
) -> "DocDBCluster":
"""
Modifies a setting for an Amazon DocumentDB cluster. You can change one or more
database configuration parameters by specifying these parameters and the new
values in the request.
Args:
model: The :py:class:`DBCluster` to update.
Keyword Args:
NewDBClusterIdentifier: The new cluster identifier for the cluster when renaming a cluster. This value is stored as
a lowercase string.
ApplyImmediately: A value that specifies whether the changes in this request and any pending changes are
asynchronously applied as soon as possible, regardless of the ``PreferredMaintenanceWindow`` setting for the
cluster. If this parameter is set to ``false``, changes to the cluster are applied during the next maintenance
window.
DBClusterParameterGroupName: The name of the cluster parameter group to use for the cluster.
VpcSecurityGroupIds: A list of virtual private cloud (VPC) security groups that the cluster will belong to.
MasterUserPassword: The password for the master database user. This password can contain any printable ASCII
character except forward slash (/), double quote ("), or the "at" symbol (@).
CloudwatchLogsExportConfiguration: The configuration setting for the log types to be enabled for export to Amazon
CloudWatch Logs for a specific instance or cluster. The ``EnableLogTypes`` and ``DisableLogTypes`` arrays determine
which logs are exported (or not exported) to CloudWatch Logs.
AllowMajorVersionUpgrade: A value that indicates whether major version upgrades are allowed.
ManageMasterUserPassword: Specifies whether to manage the master user password with Amazon Web Services Secrets
Manager. If the cluster doesn't manage the master user password with Amazon Web Services Secrets Manager, you can
turn on this management. In this case, you can't specify ``MasterUserPassword``. If the cluster already manages the
master user password with Amazon Web Services Secrets Manager, and you specify that the master user password is not
managed with Amazon Web Services Secrets Manager, then you must specify ``MasterUserPassword``. In this case, Amazon
DocumentDB deletes the secret and uses the new password for the master user specified by ``MasterUserPassword``.
MasterUserSecretKmsKeyId: The Amazon Web Services KMS key identifier to encrypt a secret that is automatically
generated and managed in Amazon Web Services Secrets Manager.
RotateMasterUserPassword: Specifies whether to rotate the secret managed by Amazon Web Services Secrets Manager for
the master user password.
"""
data = model.model_dump(exclude_none=True, by_alias=True)
args = dict(
DBClusterIdentifier=data.get("DBClusterIdentifier"),
NewDBClusterIdentifier=self.serialize(NewDBClusterIdentifier),
ApplyImmediately=self.serialize(ApplyImmediately),
BackupRetentionPeriod=data.get("BackupRetentionPeriod"),
DBClusterParameterGroupName=self.serialize(DBClusterParameterGroupName),
VpcSecurityGroupIds=self.serialize(VpcSecurityGroupIds),
Port=data.get("Port"),
MasterUserPassword=self.serialize(MasterUserPassword),
PreferredBackupWindow=data.get("PreferredBackupWindow"),
PreferredMaintenanceWindow=data.get("PreferredMaintenanceWindow"),
CloudwatchLogsExportConfiguration=self.serialize(
CloudwatchLogsExportConfiguration
),
EngineVersion=data.get("EngineVersion"),
AllowMajorVersionUpgrade=self.serialize(AllowMajorVersionUpgrade),
DeletionProtection=data.get("DeletionProtection"),
StorageType=data.get("StorageType"),
ServerlessV2ScalingConfiguration=data.get(
"ServerlessV2ScalingConfiguration"
),
ManageMasterUserPassword=self.serialize(ManageMasterUserPassword),
MasterUserSecretKmsKeyId=self.serialize(MasterUserSecretKmsKeyId),
RotateMasterUserPassword=self.serialize(RotateMasterUserPassword),
NetworkType=data.get("NetworkType"),
)
_response = self.client.modify_db_cluster(
**{k: v for k, v in args.items() if v is not None}
)
response = ModifyDBClusterResult(**_response)
self.sessionize(response.DBCluster)
return cast("DocDBCluster", response.DBCluster)
[docs] def delete(
self,
DBClusterIdentifier: str,
*,
SkipFinalSnapshot: "bool | None" = None,
FinalDBSnapshotIdentifier: "str | None" = None,
) -> "DocDBCluster":
"""
Deletes a previously provisioned cluster. When you delete a cluster, all
automated backups for that cluster are deleted and can't be recovered. Manual DB
cluster snapshots of the specified cluster are not deleted.
Args:
DBClusterIdentifier: The cluster identifier for the cluster to be deleted. This parameter isn't case sensitive.
Keyword Args:
SkipFinalSnapshot: Determines whether a final cluster snapshot is created before the cluster is deleted. If ``true``
is specified, no cluster snapshot is created. If ``false`` is specified, a cluster snapshot is created before the DB
cluster is deleted.
FinalDBSnapshotIdentifier: The cluster snapshot identifier of the new cluster snapshot created when
``SkipFinalSnapshot`` is set to ``false``.
"""
args: dict[str, Any] = dict(
DBClusterIdentifier=self.serialize(DBClusterIdentifier),
SkipFinalSnapshot=self.serialize(SkipFinalSnapshot),
FinalDBSnapshotIdentifier=self.serialize(FinalDBSnapshotIdentifier),
)
_response = self.client.delete_db_cluster(
**{k: v for k, v in args.items() if v is not None}
)
response = DeleteDBClusterResult(**_response)
return cast("DocDBCluster", response.DBCluster)
[docs] @single_docdb_cluster_include_tags
def get(self, *, DBClusterIdentifier: "str | None" = None) -> "DocDBCluster | None":
"""
Returns information about provisioned Amazon DocumentDB clusters. This API
operation supports pagination. For certain management features such as cluster
and instance lifecycle management, Amazon DocumentDB leverages operational
technology that is shared with Amazon RDS and Amazon Neptune. Use the
``filterName=engine,Values=docdb`` filter parameter to return only Amazon
DocumentDB clusters.
Keyword Args:
DBClusterIdentifier: The user-provided cluster identifier. If this parameter is specified, information from only the
specific cluster is returned. This parameter isn't case sensitive.
"""
args: dict[str, Any] = dict(
DBClusterIdentifier=self.serialize(DBClusterIdentifier)
)
_response = self.client.describe_db_clusters(
**{k: v for k, v in args.items() if v is not None}
)
response = DocDBClusterMessage(**_response)
if response and response.DBClusters:
self.sessionize(response.DBClusters[0])
return response.DBClusters[0]
return None
[docs] @multiple_docdb_cluster_include_tags
def list(
self,
*,
DBClusterIdentifier: "str | None" = None,
Filters: "builtins.list[Filter] | None" = None,
) -> PrimaryBoto3ModelQuerySet:
"""
Returns information about provisioned Amazon DocumentDB clusters. This API
operation supports pagination. For certain management features such as cluster
and instance lifecycle management, Amazon DocumentDB leverages operational
technology that is shared with Amazon RDS and Amazon Neptune. Use the
``filterName=engine,Values=docdb`` filter parameter to return only Amazon
DocumentDB clusters.
Keyword Args:
DBClusterIdentifier: The user-provided cluster identifier. If this parameter is specified, information from only the
specific cluster is returned. This parameter isn't case sensitive.
Filters: A filter that specifies one or more clusters to describe.
"""
paginator = self.client.get_paginator("describe_db_clusters")
args: dict[str, Any] = dict(
DBClusterIdentifier=self.serialize(DBClusterIdentifier),
Filters=self.serialize(Filters),
)
response_iterator = paginator.paginate(
**{k: v for k, v in args.items() if v is not None}
)
results = []
for _response in response_iterator:
if list(_response.keys()) == ["ResponseMetadata"]:
break
if "ResponseMetadata" in _response:
del _response["ResponseMetadata"]
response = DocDBClusterMessage(**_response)
if response.DBClusters:
results.extend(response.DBClusters)
else:
if getattr(response, "NextToken", None):
continue
break
self.sessionize(results)
if results and isinstance(results[0], Boto3Model):
return PrimaryBoto3ModelQuerySet(results)
return results
[docs] @single_docdb_cluster_include_tags
def failover(
self,
*,
DBClusterIdentifier: "str | None" = None,
TargetDBInstanceIdentifier: "str | None" = None,
) -> "DocDBCluster":
"""
Forces a failover for a cluster.
Keyword Args:
DBClusterIdentifier: A cluster identifier to force a failover for. This parameter is not case sensitive.
TargetDBInstanceIdentifier: The name of the instance to promote to the primary instance.
"""
args: dict[str, Any] = dict(
DBClusterIdentifier=self.serialize(DBClusterIdentifier),
TargetDBInstanceIdentifier=self.serialize(TargetDBInstanceIdentifier),
)
_response = self.client.failover_db_cluster(
**{k: v for k, v in args.items() if v is not None}
)
response = FailoverDBClusterResult(**_response)
results: "DocDBCluster" = None
if response is not None:
results = response.DBCluster
self.sessionize(results)
return cast("DocDBCluster", results)
[docs] @single_docdb_cluster_include_tags
def restore_to_point_in_time(
self,
DBClusterIdentifier: str,
SourceDBClusterIdentifier: str,
*,
RestoreType: "str | None" = None,
RestoreToTime: "datetime | None" = None,
UseLatestRestorableTime: "bool | None" = None,
Port: "int | None" = None,
DBSubnetGroupName: "str | None" = None,
VpcSecurityGroupIds: "builtins.list[str] | None" = None,
Tags: "builtins.list[Tag] | None" = None,
KmsKeyId: "str | None" = None,
EnableCloudwatchLogsExports: "builtins.list[str] | None" = None,
DeletionProtection: "bool | None" = None,
ServerlessV2ScalingConfiguration: "ServerlessV2ScalingConfiguration | None" = None,
StorageType: "str | None" = None,
NetworkType: "str | None" = None,
) -> "DocDBCluster":
"""
Restores a cluster to an arbitrary point in time. Users can restore to any point
in time before ``LatestRestorableTime`` for up to ``BackupRetentionPeriod``
days. The target cluster is created from the source cluster with the same
configuration as the original cluster, except that the new cluster is created
with the default security group.
Args:
DBClusterIdentifier: The name of the new cluster to be created.
SourceDBClusterIdentifier: The identifier of the source cluster from which to restore.
Keyword Args:
RestoreType: The type of restore to be performed. You can specify one of the following values:
RestoreToTime: The date and time to restore the cluster to.
UseLatestRestorableTime: A value that is set to ``true`` to restore the cluster to the latest restorable backup
time, and ``false`` otherwise.
Port: The port number on which the new cluster accepts connections.
DBSubnetGroupName: The subnet group name to use for the new cluster.
VpcSecurityGroupIds: A list of VPC security groups that the new cluster belongs to.
Tags: The tags to be assigned to the restored cluster.
KmsKeyId: The KMS key identifier to use when restoring an encrypted cluster from an encrypted cluster.
EnableCloudwatchLogsExports: A list of log types that must be enabled for exporting to Amazon CloudWatch Logs.
DeletionProtection: Specifies whether this cluster can be deleted. If ``DeletionProtection`` is enabled, the cluster
cannot be deleted unless it is modified and ``DeletionProtection`` is disabled. ``DeletionProtection`` protects
clusters from being accidentally deleted.
ServerlessV2ScalingConfiguration: Contains the scaling configuration of an Amazon DocumentDB Serverless cluster.
StorageType: The storage type to associate with the DB cluster.
NetworkType: The network type of the cluster.
"""
args: dict[str, Any] = dict(
DBClusterIdentifier=self.serialize(DBClusterIdentifier),
SourceDBClusterIdentifier=self.serialize(SourceDBClusterIdentifier),
RestoreType=self.serialize(RestoreType),
RestoreToTime=self.serialize(RestoreToTime),
UseLatestRestorableTime=self.serialize(UseLatestRestorableTime),
Port=self.serialize(Port),
DBSubnetGroupName=self.serialize(DBSubnetGroupName),
VpcSecurityGroupIds=self.serialize(VpcSecurityGroupIds),
Tags=self.serialize(Tags),
KmsKeyId=self.serialize(KmsKeyId),
EnableCloudwatchLogsExports=self.serialize(EnableCloudwatchLogsExports),
DeletionProtection=self.serialize(DeletionProtection),
ServerlessV2ScalingConfiguration=self.serialize(
ServerlessV2ScalingConfiguration
),
StorageType=self.serialize(StorageType),
NetworkType=self.serialize(NetworkType),
)
_response = self.client.restore_db_cluster_to_point_in_time(
**{k: v for k, v in args.items() if v is not None}
)
response = RestoreDBClusterToPointInTimeResult(**_response)
results: "DocDBCluster" = None
if response is not None:
results = response.DBCluster
self.sessionize(results)
return cast("DocDBCluster", results)
[docs] @single_docdb_cluster_include_tags
def start(self, DBClusterIdentifier: str) -> "DocDBCluster":
"""
Restarts the stopped cluster that is specified by ``DBClusterIdentifier``. For more information, see `Stopping and
Starting an Amazon DocumentDB Cluster <https://docs.aws.amazon.com/documentdb/latest/developerguide/db-cluster-stop-
start.html>`_.
Args:
DBClusterIdentifier: The identifier of the cluster to restart. Example: ``docdb-2019-05-28-15-24-52``
"""
args: dict[str, Any] = dict(
DBClusterIdentifier=self.serialize(DBClusterIdentifier)
)
_response = self.client.start_db_cluster(
**{k: v for k, v in args.items() if v is not None}
)
response = StartDBClusterResult(**_response)
results: "DocDBCluster" = None
if response is not None:
results = response.DBCluster
self.sessionize(results)
return cast("DocDBCluster", results)
[docs] @single_docdb_cluster_include_tags
def stop(self, DBClusterIdentifier: str) -> "DocDBCluster":
"""
Stops the running cluster that is specified by ``DBClusterIdentifier``. The cluster must be in the *available* state.
For more information, see `Stopping and Starting an Amazon DocumentDB
Cluster <https://docs.aws.amazon.com/documentdb/latest/developerguide/db-cluster-stop-start.html>`_.
Args:
DBClusterIdentifier: The identifier of the cluster to stop. Example: ``docdb-2019-05-28-15-24-52``
"""
args: dict[str, Any] = dict(
DBClusterIdentifier=self.serialize(DBClusterIdentifier)
)
_response = self.client.stop_db_cluster(
**{k: v for k, v in args.items() if v is not None}
)
response = StopDBClusterResult(**_response)
results: "DocDBCluster" = None
if response is not None:
results = response.DBCluster
self.sessionize(results)
return cast("DocDBCluster", results)
[docs]class DocDBInstanceManager(Boto3ModelManager):
service_name: str = "docdb"
[docs] @single_docdb_instance_include_tags
def create(self, model: "DocDBInstance") -> "DocDBInstance":
"""
Creates a new instance.
Args:
model: The :py:class:`DBInstance` to create.
"""
data = model.model_dump(exclude_none=True, by_alias=True)
args = dict(
DBInstanceIdentifier=data.get("DBInstanceIdentifier"),
DBInstanceClass=data.get("DBInstanceClass"),
Engine=data.get("Engine"),
DBClusterIdentifier=data.get("DBClusterIdentifier"),
AvailabilityZone=data.get("AvailabilityZone"),
PreferredMaintenanceWindow=data.get("PreferredMaintenanceWindow"),
AutoMinorVersionUpgrade=data.get("AutoMinorVersionUpgrade"),
Tags=data.get("Tags"),
CopyTagsToSnapshot=data.get("CopyTagsToSnapshot"),
PromotionTier=data.get("PromotionTier"),
EnablePerformanceInsights=data.get("PerformanceInsightsEnabled"),
PerformanceInsightsKMSKeyId=data.get("PerformanceInsightsKMSKeyId"),
CACertificateIdentifier=data.get("CACertificateIdentifier"),
)
_response = self.client.create_db_instance(
**{k: v for k, v in args.items() if v is not None}
)
response = DocDBCreateDBInstanceResult(**_response)
self.sessionize(response.DBInstance)
return cast("DocDBInstance", response.DBInstance)
[docs] @single_docdb_instance_include_tags
def update(
self,
model: "DocDBInstance",
ApplyImmediately: "bool | None" = None,
NewDBInstanceIdentifier: "str | None" = None,
CertificateRotationRestart: "bool | None" = None,
) -> "DocDBInstance":
"""
Modifies settings for an instance. You can change one or more database
configuration parameters by specifying these parameters and the new values in
the request.
Args:
model: The :py:class:`DBInstance` to update.
Keyword Args:
ApplyImmediately: Specifies whether the modifications in this request and any pending modifications are
asynchronously applied as soon as possible, regardless of the ``PreferredMaintenanceWindow`` setting for the
instance.
NewDBInstanceIdentifier: The new instance identifier for the instance when renaming an instance. When you change the
instance identifier, an instance reboot occurs immediately if you set ``Apply Immediately`` to ``true``. It occurs
during the next maintenance window if you set ``Apply Immediately`` to ``false``. This value is stored as a
lowercase string.
CertificateRotationRestart: Specifies whether the DB instance is restarted when you rotate your SSL/TLS certificate.
"""
data = model.model_dump(exclude_none=True, by_alias=True)
args = dict(
DBInstanceIdentifier=data.get("DBInstanceIdentifier"),
DBInstanceClass=data.get("DBInstanceClass"),
ApplyImmediately=self.serialize(ApplyImmediately),
PreferredMaintenanceWindow=data.get("PreferredMaintenanceWindow"),
AutoMinorVersionUpgrade=data.get("AutoMinorVersionUpgrade"),
NewDBInstanceIdentifier=self.serialize(NewDBInstanceIdentifier),
CACertificateIdentifier=data.get("CACertificateIdentifier"),
CopyTagsToSnapshot=data.get("CopyTagsToSnapshot"),
PromotionTier=data.get("PromotionTier"),
EnablePerformanceInsights=data.get("PerformanceInsightsEnabled"),
PerformanceInsightsKMSKeyId=data.get("PerformanceInsightsKMSKeyId"),
CertificateRotationRestart=self.serialize(CertificateRotationRestart),
)
_response = self.client.modify_db_instance(
**{k: v for k, v in args.items() if v is not None}
)
response = DocDBModifyDBInstanceResult(**_response)
self.sessionize(response.DBInstance)
return cast("DocDBInstance", response.DBInstance)
[docs] def delete(self, DBInstanceIdentifier: str) -> "DocDBInstance":
"""
Deletes a previously provisioned instance.
Args:
DBInstanceIdentifier: The instance identifier for the instance to be deleted. This parameter isn't case sensitive.
"""
args: dict[str, Any] = dict(
DBInstanceIdentifier=self.serialize(DBInstanceIdentifier)
)
_response = self.client.delete_db_instance(
**{k: v for k, v in args.items() if v is not None}
)
response = DocDBDeleteDBInstanceResult(**_response)
return cast("DocDBInstance", response.DBInstance)
[docs] @multiple_docdb_instance_include_tags
def list(
self,
*,
DBInstanceIdentifier: "str | None" = None,
Filters: "builtins.list[Filter] | None" = None,
) -> PrimaryBoto3ModelQuerySet:
"""
Returns information about provisioned Amazon DocumentDB instances. This API
supports pagination.
Keyword Args:
DBInstanceIdentifier: The user-provided instance identifier. If this parameter is specified, information from only
the specific instance is returned. This parameter isn't case sensitive.
Filters: A filter that specifies one or more instances to describe.
"""
paginator = self.client.get_paginator("describe_db_instances")
args: dict[str, Any] = dict(
DBInstanceIdentifier=self.serialize(DBInstanceIdentifier),
Filters=self.serialize(Filters),
)
response_iterator = paginator.paginate(
**{k: v for k, v in args.items() if v is not None}
)
results = []
for _response in response_iterator:
if list(_response.keys()) == ["ResponseMetadata"]:
break
if "ResponseMetadata" in _response:
del _response["ResponseMetadata"]
response = DocDBInstanceMessage(**_response)
if response.DBInstances:
results.extend(response.DBInstances)
else:
if getattr(response, "NextToken", None):
continue
break
self.sessionize(results)
if results and isinstance(results[0], Boto3Model):
return PrimaryBoto3ModelQuerySet(results)
return results
[docs] @single_docdb_instance_include_tags
def get(self, DBInstanceIdentifier: str) -> "DocDBInstance | None":
"""
Returns information about provisioned Amazon DocumentDB instances. This API
supports pagination.
Args:
DBInstanceIdentifier: The user-provided instance identifier. If this parameter is specified, information from only
the specific instance is returned. This parameter isn't case sensitive.
"""
args: dict[str, Any] = dict(
DBInstanceIdentifier=self.serialize(DBInstanceIdentifier)
)
_response = self.client.describe_db_instances(
**{k: v for k, v in args.items() if v is not None}
)
response = DocDBInstanceMessage(**_response)
if response and response.DBInstances:
self.sessionize(response.DBInstances[0])
return response.DBInstances[0]
return None
[docs] @single_docdb_instance_include_tags
def reboot(
self, DBInstanceIdentifier: str, *, ForceFailover: "bool | None" = None
) -> "DocDBInstance":
"""
You might need to reboot your instance, usually for maintenance reasons. For
example, if you make certain changes, or if you change the cluster parameter
group that is associated with the instance, you must reboot the instance for the
changes to take effect.
Args:
DBInstanceIdentifier: The instance identifier. This parameter is stored as a lowercase string.
Keyword Args:
ForceFailover: When ``true``, the reboot is conducted through a Multi-AZ failover.
"""
args: dict[str, Any] = dict(
DBInstanceIdentifier=self.serialize(DBInstanceIdentifier),
ForceFailover=self.serialize(ForceFailover),
)
_response = self.client.reboot_db_instance(
**{k: v for k, v in args.items() if v is not None}
)
response = RebootDBInstanceResult(**_response)
results: "DocDBInstance" = None
if response is not None:
results = response.DBInstance
self.sessionize(results)
return cast("DocDBInstance", results)
[docs]class DocDBSubnetGroupManager(Boto3ModelManager):
service_name: str = "docdb"
[docs] @single_docdb_subnet_group_include_tags
def create(
self,
model: "DocDBSubnetGroup",
SubnetIds: "builtins.list[str]",
Tags: "builtins.list[Tag] | None" = None,
) -> "DocDBSubnetGroup":
"""
Creates a new subnet group. subnet groups must contain at least one subnet in at
least two Availability Zones in the Amazon Web Services Region.
Args:
model: The :py:class:`DBSubnetGroup` to create.
SubnetIds: The Amazon EC2 subnet IDs for the subnet group.
Keyword Args:
Tags: The tags to be assigned to the subnet group.
"""
data = model.model_dump(exclude_none=True, by_alias=True)
args = dict(
DBSubnetGroupName=data.get("DBSubnetGroupName"),
DBSubnetGroupDescription=data.get("DBSubnetGroupDescription"),
SubnetIds=self.serialize(SubnetIds),
Tags=self.serialize(Tags),
)
_response = self.client.create_db_subnet_group(
**{k: v for k, v in args.items() if v is not None}
)
response = DocDBCreateDBSubnetGroupResult(**_response)
self.sessionize(response.DBSubnetGroup)
return cast("DocDBSubnetGroup", response.DBSubnetGroup)
[docs] @single_docdb_subnet_group_include_tags
def update(
self, model: "DocDBSubnetGroup", SubnetIds: "builtins.list[str]"
) -> "DocDBSubnetGroup":
"""
Modifies an existing subnet group. subnet groups must contain at least one
subnet in at least two Availability Zones in the Amazon Web Services Region.
Args:
model: The :py:class:`DBSubnetGroup` to update.
SubnetIds: The Amazon EC2 subnet IDs for the subnet group.
"""
data = model.model_dump(exclude_none=True, by_alias=True)
args = dict(
DBSubnetGroupName=data.get("DBSubnetGroupName"),
SubnetIds=self.serialize(SubnetIds),
DBSubnetGroupDescription=data.get("DBSubnetGroupDescription"),
)
_response = self.client.modify_db_subnet_group(
**{k: v for k, v in args.items() if v is not None}
)
response = DocDBModifyDBSubnetGroupResult(**_response)
self.sessionize(response.DBSubnetGroup)
return cast("DocDBSubnetGroup", response.DBSubnetGroup)
[docs] def delete(self, DBSubnetGroupName: str) -> None:
"""
Deletes a subnet group.
Args:
DBSubnetGroupName: The name of the database subnet group to delete.
"""
args: dict[str, Any] = dict(DBSubnetGroupName=self.serialize(DBSubnetGroupName))
self.client.delete_db_subnet_group(
**{k: v for k, v in args.items() if v is not None}
)
[docs] @single_docdb_subnet_group_include_tags
def get(self, DBSubnetGroupName: str) -> "DocDBSubnetGroup | None":
"""
Returns a list of ``DBSubnetGroup`` descriptions. If a ``DBSubnetGroupName`` is
specified, the list will contain only the descriptions of the specified
``DBSubnetGroup``.
Args:
DBSubnetGroupName: The name of the subnet group to return details for.
"""
args: dict[str, Any] = dict(DBSubnetGroupName=self.serialize(DBSubnetGroupName))
_response = self.client.describe_db_subnet_groups(
**{k: v for k, v in args.items() if v is not None}
)
response = DocDBSubnetGroupMessage(**_response)
if response and response.DBSubnetGroups:
self.sessionize(response.DBSubnetGroups[0])
return response.DBSubnetGroups[0]
return None
[docs] @multiple_docdb_subnet_group_include_tags
def list(
self,
*,
DBSubnetGroupName: "str | None" = None,
Filters: "builtins.list[Filter] | None" = None,
) -> PrimaryBoto3ModelQuerySet:
"""
Returns a list of ``DBSubnetGroup`` descriptions. If a ``DBSubnetGroupName`` is
specified, the list will contain only the descriptions of the specified
``DBSubnetGroup``.
Keyword Args:
DBSubnetGroupName: The name of the subnet group to return details for.
Filters: This parameter is not currently supported.
"""
paginator = self.client.get_paginator("describe_db_subnet_groups")
args: dict[str, Any] = dict(
DBSubnetGroupName=self.serialize(DBSubnetGroupName),
Filters=self.serialize(Filters),
)
response_iterator = paginator.paginate(
**{k: v for k, v in args.items() if v is not None}
)
results = []
for _response in response_iterator:
if list(_response.keys()) == ["ResponseMetadata"]:
break
if "ResponseMetadata" in _response:
del _response["ResponseMetadata"]
response = DocDBSubnetGroupMessage(**_response)
if response.DBSubnetGroups:
results.extend(response.DBSubnetGroups)
else:
if getattr(response, "NextToken", None):
continue
break
self.sessionize(results)
if results and isinstance(results[0], Boto3Model):
return PrimaryBoto3ModelQuerySet(results)
return results
# ==============
# Service Models
# ==============
[docs]class DBClusterMember(Boto3Model):
"""
Contains information about an instance that is part of a cluster.
"""
DBInstanceIdentifier: "str | None" = None
"""
Specifies the instance identifier for this member of the cluster.
"""
IsClusterWriter: "bool | None" = None
"""
A value that is ``true`` if the cluster member is the primary instance for the
cluster and ``false`` otherwise.
"""
DBClusterParameterGroupStatus: "str | None" = None
"""
Specifies the status of the cluster parameter group for this member of the DB
cluster.
"""
PromotionTier: "int | None" = None
"""
A value that specifies the order in which an Amazon DocumentDB replica is promoted
to the primary instance after a failure of the existing primary instance.
"""
[docs]class VpcSecurityGroupMembership(Boto3Model):
"""
Used as a response element for queries on virtual private cloud (VPC) security group
membership.
"""
VpcSecurityGroupId: "str | None" = None
"""
The name of the VPC security group.
"""
Status: "str | None" = None
"""
The status of the VPC security group.
"""
[docs]class DBClusterRole(Boto3Model):
"""
Describes an Identity and Access Management (IAM) role that is associated with a
cluster.
"""
RoleArn: "str | None" = None
"""
The Amazon Resource Name (ARN) of the IAMrole that is associated with the DB
cluster.
"""
Status: "str | None" = None
"""
Describes the state of association between the IAMrole and the cluster.
The ``Status`` property returns one of the
following values:
"""
[docs]class ServerlessV2ScalingConfigurationInfo(Boto3Model):
"""
Retrieves the scaling configuration for an Amazon DocumentDB Serverless cluster.
"""
MinCapacity: "float | None" = None
"""
The minimum number of Amazon DocumentDB capacity units (DCUs) for an instance in an
Amazon DocumentDB Serverless cluster.
You can specify DCU values in half-step increments, such as 8, 8.5, 9, and so on.
"""
MaxCapacity: "float | None" = None
"""
The maximum number of Amazon DocumentDB capacity units (DCUs) for an instance in an
Amazon DocumentDB Serverless cluster.
You can specify DCU values in half-step increments, such as 32, 32.5, 33, and so on.
"""
[docs]class ClusterMasterUserSecret(Boto3Model):
"""
Contains the secret managed by Amazon DocumentDB in Amazon Web Services Secrets
Manager for the master user password.
"""
SecretArn: "str | None" = None
"""
The Amazon Resource Name (ARN) of the secret.
"""
SecretStatus: "str | None" = None
"""
The status of the secret.
"""
KmsKeyId: "str | None" = None
"""
The Amazon Web Services KMS key identifier that is used to encrypt the secret.
"""
[docs]class DocDBCluster(TagsDictMixin, DocDBClusterModelMixin, PrimaryBoto3Model):
"""
Detailed information about a cluster.
"""
tag_class: ClassVar[type[Boto3Model]] = Tag
manager_class: ClassVar[type[Boto3ModelManager]] = DocDBClusterManager
DBSubnetGroup: "str | None" = None
"""
Specifies information on the subnet group that is associated with the cluster,
including the name, description, and subnets in the subnet group.
"""
AvailabilityZones: "builtins.list[str] | None" = Field(default_factory=list)
"""
Provides the list of Amazon EC2 Availability Zones that instances in the cluster can
be created in.
"""
BackupRetentionPeriod: "int | None" = None
"""
Specifies the number of days for which automatic snapshots are retained.
"""
DBClusterIdentifier: "str | None" = None
"""
Contains a user-supplied cluster identifier.
This identifier is the unique key that identifies a cluster.
"""
DBClusterParameterGroup: str = Field(default=None, frozen=True)
"""
Specifies the name of the cluster parameter group for the cluster.
"""
Status: str = Field(default=None, frozen=True)
"""
Specifies the current state of this cluster.
"""
PercentProgress: str = Field(default=None, frozen=True)
"""
Specifies the progress of the operation as a percentage.
"""
EarliestRestorableTime: datetime = Field(default=None, frozen=True)
"""
The earliest time to which a database can be restored with point-in-time restore.
"""
Endpoint: str = Field(default=None, frozen=True)
"""
Specifies the connection endpoint for the primary instance of the cluster.
"""
ReaderEndpoint: str = Field(default=None, frozen=True)
"""
The reader endpoint for the cluster.
The reader endpoint for a cluster load balances connections across the Amazon
DocumentDB replicas that are available in a cluster. As clients request new
connections to the reader endpoint, Amazon DocumentDB distributes the connection
requests among the Amazon DocumentDB replicas in the cluster. This functionality can
help balance your read workload across multiple Amazon DocumentDB replicas in your
cluster.
"""
MultiAZ: bool = Field(default=None, frozen=True)
"""
Specifies whether the cluster has instances in multiple Availability Zones.
"""
Engine: "str | None" = None
"""
Provides the name of the database engine to be used for this cluster.
"""
EngineVersion: "str | None" = None
"""
Indicates the database engine version.
"""
LatestRestorableTime: datetime = Field(default=None, frozen=True)
"""
Specifies the latest time to which a database can be restored with point-in-time
restore.
"""
Port: "int | None" = None
"""
Specifies the port that the database engine is listening on.
"""
MasterUsername: "str | None" = None
"""
Contains the master user name for the cluster.
"""
PreferredBackupWindow: "str | None" = None
"""
Specifies the daily time range during which automated backups are created if
automated backups are enabled, as determined by the ``BackupRetentionPeriod``.
"""
PreferredMaintenanceWindow: "str | None" = None
"""
Specifies the weekly time range during which system maintenance can occur, in
Universal Coordinated Time (UTC).
"""
ReplicationSourceIdentifier: str = Field(default=None, frozen=True)
"""
Contains the identifier of the source cluster if this cluster is a secondary
cluster.
"""
ReadReplicaIdentifiers: "builtins.list[str]" = Field(
default_factory=list, frozen=True
)
"""
Contains one or more identifiers of the secondary clusters that are associated with
this cluster.
"""
DBClusterMembers: "builtins.list[DBClusterMember]" = Field(
default_factory=list, frozen=True
)
"""
Provides the list of instances that make up the cluster.
"""
VpcSecurityGroups: "builtins.list[VpcSecurityGroupMembership]" = Field(
default_factory=list, frozen=True
)
"""
Provides a list of virtual private cloud (VPC) security groups that the cluster
belongs to.
"""
HostedZoneId: str = Field(default=None, frozen=True)
"""
Specifies the ID that Amazon Route 53 assigns when you create a hosted zone.
"""
StorageEncrypted: "bool | None" = None
"""
Specifies whether the cluster is encrypted.
"""
KmsKeyId: "str | None" = None
"""
If ``StorageEncrypted`` is ``true``, the KMS key identifier for the encrypted
cluster.
"""
DbClusterResourceId: str = Field(default=None, frozen=True)
"""
The Amazon Web Services Region-unique, immutable identifier for the cluster.
This identifier is found in CloudTrail log entries whenever the KMS key for the
cluster is accessed.
"""
DBClusterArn: str = Field(default=None, frozen=True)
"""
The Amazon Resource Name (ARN) for the cluster.
"""
AssociatedRoles: "builtins.list[DBClusterRole]" = Field(
default_factory=list, frozen=True
)
"""
Provides a list of the Identity and Access Management (IAM) roles that are
associated with the cluster.
(IAM) roles that are associated with a cluster grant permission for the cluster to
access other Amazon Web Services services on your behalf.
"""
CloneGroupId: str = Field(default=None, frozen=True)
"""
Identifies the clone group to which the DB cluster is associated.
"""
ClusterCreateTime: datetime = Field(default=None, frozen=True)
"""
Specifies the time when the cluster was created, in Universal Coordinated Time
(UTC).
"""
EnabledCloudwatchLogsExports: "builtins.list[str]" = Field(
default_factory=list, frozen=True
)
"""
A list of log types that this cluster is configured to export to Amazon CloudWatch
Logs.
"""
DeletionProtection: "bool | None" = None
"""
Specifies whether this cluster can be deleted.
If ``DeletionProtection`` is enabled, the cluster cannot be deleted
unless it is modified and ``DeletionProtection`` is disabled. ``DeletionProtection`` protects clusters from being
accidentally deleted.
"""
IOOptimizedNextAllowedModificationTime: datetime = Field(default=None, frozen=True)
"""
The next time you can modify the Amazon DocumentDB cluster to use the iopt1 storage
type.
"""
StorageType: "str | None" = None
"""
Storage type associated with your cluster.
"""
ServerlessV2ScalingConfiguration: "ServerlessV2ScalingConfigurationInfo | None" = (
None
)
"""
The scaling configuration of an Amazon DocumentDB Serverless cluster.
"""
MasterUserSecret: ClusterMasterUserSecret = Field(default=None, frozen=True)
"""
The secret managed by Amazon DocumentDB in Amazon Web Services Secrets Manager for
the master user password.
"""
NetworkType: "str | None" = None
"""
The network type of the cluster.
"""
Tags: "builtins.list[Tag] | None" = Field(default_factory=list)
"""
The tags associated with the DocumentDB cluster.
"""
@property
def pk(self) -> str | None:
"""
Return the primary key of the model. This is the value of the
:py:attr:`DBClusterIdentifier` attribute.
Returns:
The primary key of the model instance.
"""
return self.DBClusterIdentifier
@property
def arn(self) -> str | None:
"""
Return the ARN of the model. This is the value of the :py:attr:`DBClusterArn`
attribute.
Returns:
The ARN of the model instance.
"""
return self.DBClusterArn
@property
def name(self) -> str | None:
"""
Return the name of the model. This is the value of the
:py:attr:`DBClusterIdentifier` attribute.
Returns:
The name of the model instance.
"""
return self.DBClusterIdentifier
def __hash__(self) -> int:
"""
Return the hash of the model.
This is the value of the
:py:attr:`DBClusterIdentifier` attribute.
"""
return hash(self.DBClusterIdentifier)
@cached_property
def subnet_group(self) -> "DocDBSubnetGroup | None":
"""
Return the subnet group that this DB instance is in, if any.
.. note::
The output of this property is cached on the model instance, so
calling this multiple times will not result in multiple calls to the
AWS API. If you need a fresh copy of the data, you can re-get the
model instance from the manager.
"""
try:
pk = OrderedDict(
{
"DBSubnetGroupName": self.DBSubnetGroup,
}
)
except AttributeError:
return None
return DocDBSubnetGroup.objects.using(self.session).get(**pk) # type: ignore[arg-type]
@cached_property
def security_groups(self) -> "list[SecurityGroup] | None":
"""
Return the security groups that this DB instance is in, if any.
.. note::
The output of this property is cached on the model instance, so
calling this multiple times will not result in multiple calls to the
AWS API. If you need a fresh copy of the data, you can re-get the
model instance from the manager.
"""
try:
pk = OrderedDict(
{
"GroupIds": [
g.VpcSecurityGroupId
for g in self.VpcSecurityGroups
if g.VpcSecurityGroupId
],
}
)
except AttributeError:
return []
return SecurityGroup.objects.using(self.session).list(**pk) # type: ignore[arg-type]
@cached_property
def vpc(self) -> "Vpc | None":
"""
Return the VPC that this DB instance is in, if any.
.. note::
The output of this property is cached on the model instance, so
calling this multiple times will not result in multiple calls to the
AWS API. If you need a fresh copy of the data, you can re-get the
model instance from the manager.
"""
try:
pk = OrderedDict(
{
"VpcId": self.subnet_group.VpcId,
}
)
except AttributeError:
return None
return Vpc.objects.using(self.session).get(**pk) # type: ignore[arg-type]
@cached_property
def kms_key(self) -> "KMSKey | None":
"""
Return the KMS key that is used to encrypt the data in the DB cluster.
.. note::
The output of this property is cached on the model instance, so
calling this multiple times will not result in multiple calls to the
AWS API. If you need a fresh copy of the data, you can re-get the
model instance from the manager.
"""
try:
pk = OrderedDict(
{
"KeyId": self.KmsKeyId,
}
)
except AttributeError:
return None
return KMSKey.objects.using(self.session).get(**pk) # type: ignore[arg-type]
[docs] def start(self) -> "DocDBCluster":
"""
Restarts the stopped cluster that is specified by ``DBClusterIdentifier``.
For more information, see
`Stopping and Starting an Amazon DocumentDB Cluster <https://docs.aws.amazon.com/documentdb/latest/developerguide/db-cluster-stop-
start.html>`_.
"""
return (
cast("DocDBClusterManager", self.objects) # type: ignore[attr-defined]
.using(self.session)
.start(
self.DBClusterIdentifier,
)
)
[docs] def stop(self) -> "DocDBCluster":
"""
Stops the running cluster that is specified by ``DBClusterIdentifier``.
The cluster must be in the *available* state. For more information, see
`Stopping and Starting an Amazon DocumentDB Cluster <https://docs.aws.amazon.com/documentdb/latest/developerguide/db-cluster-stop-start.html>`_.
"""
return (
cast("DocDBClusterManager", self.objects) # type: ignore[attr-defined]
.using(self.session)
.stop(
self.DBClusterIdentifier,
)
)
[docs] def failover(
self, TargetDBInstanceIdentifier: "str | None" = None
) -> "DocDBCluster":
"""
Forces a failover for a cluster.
Keyword Args:
TargetDBInstanceIdentifier: The name of the instance to promote to the primary instance.
"""
return (
cast("DocDBClusterManager", self.objects) # type: ignore[attr-defined]
.using(self.session)
.failover(
self.DBClusterIdentifier,
TargetDBInstanceIdentifier=TargetDBInstanceIdentifier,
)
)
[docs]class DocDBEndpoint(Boto3Model):
"""
Network information for accessing a cluster or instance.
Client programs must specify a valid endpoint to access these Amazon DocumentDB
resources.
"""
Address: "str | None" = None
"""
Specifies the DNS address of the instance.
"""
Port: "int | None" = None
"""
Specifies the port that the database engine is listening on.
"""
HostedZoneId: "str | None" = None
"""
Specifies the ID that Amazon Route 53 assigns when you create a hosted zone.
"""
[docs]class DocDBPendingCloudwatchLogsExports(Boto3Model):
"""
A list of the log types whose configuration is still pending.
These log types are in the process of being activated or deactivated.
"""
LogTypesToEnable: "builtins.list[str] | None" = Field(default_factory=list)
"""
Log types that are in the process of being deactivated.
After they are deactivated, these log types aren't exported to CloudWatch Logs.
"""
LogTypesToDisable: "builtins.list[str] | None" = Field(default_factory=list)
"""
Log types that are in the process of being enabled.
After they are enabled, these log types are exported to Amazon CloudWatch Logs.
"""
[docs]class DocDBPendingModifiedValues(Boto3Model):
"""
One or more modified settings for an instance.
These modified settings have been requested, but haven't been applied yet.
"""
DBInstanceClass: "str | None" = None
"""
Contains the new ``DBInstanceClass`` for the instance that will be applied or is
currently being applied.
"""
AllocatedStorage: "int | None" = None
"""
Contains the new ``AllocatedStorage`` size for then instance that will be applied or
is currently being applied.
"""
MasterUserPassword: "str | None" = None
"""
Contains the pending or currently in-progress change of the master credentials for
the instance.
"""
Port: "int | None" = None
"""
Specifies the pending port for the instance.
"""
BackupRetentionPeriod: "int | None" = None
"""
Specifies the pending number of days for which automated backups are retained.
"""
MultiAZ: "bool | None" = None
"""
Indicates that the Single-AZ instance is to change to a Multi-AZ deployment.
"""
EngineVersion: "str | None" = None
"""
Indicates the database engine version.
"""
LicenseModel: "str | None" = None
"""
The license model for the instance.
"""
Iops: "int | None" = None
"""
Specifies the new Provisioned IOPS value for the instance that will be applied or is
currently being applied.
"""
DBInstanceIdentifier: "str | None" = None
"""
Contains the new ``DBInstanceIdentifier`` for the instance that will be applied or
is currently being applied.
"""
StorageType: "str | None" = None
"""
Specifies the storage type to be associated with the instance.
"""
CACertificateIdentifier: "str | None" = None
"""
Specifies the identifier of the certificate authority (CA) certificate for the DB
instance.
"""
DBSubnetGroupName: "str | None" = None
"""
The new subnet group for the instance.
"""
PendingCloudwatchLogsExports: "DocDBPendingCloudwatchLogsExports | None" = None
"""
A list of the log types whose configuration is still pending.
These log types are in the process of being activated or deactivated.
"""
[docs]class DBInstanceStatusInfo(Boto3Model):
"""
Provides a list of status information for an instance.
"""
StatusType: "str | None" = None
"""
This value is currently "``read replication``.".
"""
Normal: "bool | None" = None
"""
A Boolean value that is ``true`` if the instance is operating normally, or ``false``
if the instance is in an error state.
"""
Status: "str | None" = None
"""
Status of the instance.
For a ``StatusType`` of read replica, the values can be ``replicating``, error, ``stopped``, or
``terminated``.
"""
Message: "str | None" = None
"""
Details of the error if there is an error for the instance.
If the instance is not in an error state, this value is blank.
"""
[docs]class DocDBCertificateDetails(Boto3Model):
"""
Returns the details of the DB instance’s server certificate.
For more information, see
`Updating Your Amazon DocumentDB TLS Certificates <https://docs.aws.amazon.com/documentdb/latest/developerguide/ca_cert_rotation.html>`_
and
`Encrypting Data in Transit <https://docs.aws.amazon.com/documentdb/latest/developerguide/security.encryption.ssl.html>`_
in the *Amazon
DocumentDB Developer Guide*.
"""
CAIdentifier: "str | None" = None
"""
The CA identifier of the CA certificate used for the DB instance's server
certificate.
"""
ValidTill: "datetime | None" = None
"""
The expiration date of the DB instance’s server certificate.
"""
[docs]class DocDBInstance(TagsDictMixin, PrimaryBoto3Model):
"""
Detailed information about an instance.
"""
tag_class: ClassVar[type[Boto3Model]] = Tag
manager_class: ClassVar[type[Boto3ModelManager]] = DocDBInstanceManager
DBSubnetGroup: "DocDBSubnetGroup | None" = None
"""
Specifies information on the subnet group that is associated with the instance,
including the name, description, and subnets in the subnet group.
"""
DBInstanceIdentifier: "str | None" = None
"""
Contains a user-provided database identifier.
This identifier is the unique key that identifies an instance.
"""
DBInstanceClass: "str | None" = None
"""
Contains the name of the compute and memory capacity class of the instance.
"""
Engine: "str | None" = None
"""
Provides the name of the database engine to be used for this instance.
"""
DBInstanceStatus: str = Field(default=None, frozen=True)
"""
Specifies the current state of this database.
"""
Endpoint: DocDBEndpoint = Field(default=None, frozen=True)
"""
Specifies the connection endpoint.
"""
InstanceCreateTime: datetime = Field(default=None, frozen=True)
"""
Provides the date and time that the instance was created.
"""
PreferredBackupWindow: str = Field(default=None, frozen=True)
"""
Specifies the daily time range during which automated backups are created if
automated backups are enabled, as determined by the ``BackupRetentionPeriod``.
"""
BackupRetentionPeriod: int = Field(default=None, frozen=True)
"""
Specifies the number of days for which automatic snapshots are retained.
"""
VpcSecurityGroups: "builtins.list[VpcSecurityGroupMembership]" = Field(
default_factory=list, frozen=True
)
"""
Provides a list of VPC security group elements that the instance belongs to.
"""
AvailabilityZone: "str | None" = None
"""
Specifies the name of the Availability Zone that the instance is located in.
"""
PreferredMaintenanceWindow: "str | None" = None
"""
Specifies the weekly time range during which system maintenance can occur, in
Universal Coordinated Time (UTC).
"""
PendingModifiedValues: DocDBPendingModifiedValues = Field(default=None, frozen=True)
"""
Specifies that changes to the instance are pending.
This element is included only when changes are pending. Specific changes are
identified by subelements.
"""
LatestRestorableTime: datetime = Field(default=None, frozen=True)
"""
Specifies the latest time to which a database can be restored with point-in-time
restore.
"""
EngineVersion: str = Field(default=None, frozen=True)
"""
Indicates the database engine version.
"""
AutoMinorVersionUpgrade: "bool | None" = None
"""
Does not apply.
This parameter does not apply to Amazon DocumentDB. Amazon DocumentDB does not
perform minor version upgrades regardless of the value set.
"""
PubliclyAccessible: bool = Field(default=None, frozen=True)
"""
Not supported.
Amazon DocumentDB does not currently support public endpoints. The value of ``PubliclyAccessible`` is
always ``false``.
"""
StatusInfos: "builtins.list[DBInstanceStatusInfo]" = Field(
default_factory=list, frozen=True
)
"""
The status of a read replica.
If the instance is not a read replica, this is blank.
"""
DBClusterIdentifier: "str | None" = None
"""
Contains the name of the cluster that the instance is a member of if the instance is
a member of a cluster.
"""
StorageEncrypted: bool = Field(default=None, frozen=True)
"""
Specifies whether or not the instance is encrypted.
"""
KmsKeyId: str = Field(default=None, frozen=True)
"""
If ``StorageEncrypted`` is ``true``, the KMS key identifier for the encrypted
instance.
"""
DbiResourceId: str = Field(default=None, frozen=True)
"""
The Amazon Web Services Region-unique, immutable identifier for the instance.
This identifier is found in CloudTrail log entries whenever the KMS key for the
instance is accessed.
"""
CACertificateIdentifier: "str | None" = None
"""
The identifier of the CA certificate for this DB instance.
"""
CopyTagsToSnapshot: "bool | None" = None
"""
A value that indicates whether to copy tags from the DB instance to snapshots of the
DB instance.
By default, tags are not copied.
"""
PromotionTier: "int | None" = None
"""
A value that specifies the order in which an Amazon DocumentDB replica is promoted
to the primary instance after a failure of the existing primary instance.
"""
DBInstanceArn: str = Field(default=None, frozen=True)
"""
The Amazon Resource Name (ARN) for the instance.
"""
EnabledCloudwatchLogsExports: "builtins.list[str]" = Field(
default_factory=list, frozen=True
)
"""
A list of log types that this instance is configured to export to CloudWatch Logs.
"""
CertificateDetails: DocDBCertificateDetails = Field(default=None, frozen=True)
"""
The details of the DB instance's server certificate.
"""
PerformanceInsightsEnabled: bool = Field(default=None, frozen=True)
"""
Set to ``true`` if Amazon RDS Performance Insights is enabled for the DB instance,
and otherwise ``false``.
"""
PerformanceInsightsKMSKeyId: "str | None" = None
"""
The KMS key identifier for encryption of Performance Insights data.
The KMS key ID is the Amazon Resource Name (ARN), KMS key identifier, or the KMS key
alias for the KMS encryption key.
"""
Tags: "builtins.list[Tag] | None" = Field(default_factory=list)
"""
The tags associated with the log group.
"""
@property
def pk(self) -> str | None:
"""
Return the primary key of the model. This is the value of the
:py:attr:`DBInstanceIdentifier` attribute.
Returns:
The primary key of the model instance.
"""
return self.DBInstanceIdentifier
@property
def arn(self) -> str | None:
"""
Return the ARN of the model. This is the value of the :py:attr:`DBInstanceArn`
attribute.
Returns:
The ARN of the model instance.
"""
return self.DBInstanceArn
@property
def name(self) -> str | None:
"""
Return the name of the model. This is the value of the
:py:attr:`DBInstanceIdentifier` attribute.
Returns:
The name of the model instance.
"""
return self.DBInstanceIdentifier
def __hash__(self) -> int:
"""
Return the hash of the model.
This is the value of the
:py:attr:`DBInstanceIdentifier` attribute.
"""
return hash(self.DBInstanceIdentifier)
@cached_property
def subnet_group(self) -> "DocDBSubnetGroup | None":
"""
Return the subnet group that this DB instance is in, if any.
.. note::
The output of this property is cached on the model instance, so
calling this multiple times will not result in multiple calls to the
AWS API. If you need a fresh copy of the data, you can re-get the
model instance from the manager.
"""
try:
pk = OrderedDict(
{
"DBSubnetGroupName": self.DBSubnetGroup,
}
)
except AttributeError:
return None
return DocDBSubnetGroup.objects.using(self.session).get(**pk) # type: ignore[arg-type]
@cached_property
def security_groups(self) -> "list[SecurityGroup] | None":
"""
Return the security groups that this DB instance is in, if any.
.. note::
The output of this property is cached on the model instance, so
calling this multiple times will not result in multiple calls to the
AWS API. If you need a fresh copy of the data, you can re-get the
model instance from the manager.
"""
try:
pk = OrderedDict(
{
"GroupIds": [
g.VpcSecurityGroupId
for g in self.VpcSecurityGroups
if g.VpcSecurityGroupId
],
}
)
except AttributeError:
return []
return SecurityGroup.objects.using(self.session).list(**pk) # type: ignore[arg-type]
@cached_property
def vpc(self) -> "Vpc | None":
"""
Return the VPC that this DB instance is in, if any.
.. note::
The output of this property is cached on the model instance, so
calling this multiple times will not result in multiple calls to the
AWS API. If you need a fresh copy of the data, you can re-get the
model instance from the manager.
"""
try:
pk = OrderedDict(
{
"VpcId": self.subnet_group.VpcId,
}
)
except AttributeError:
return None
return Vpc.objects.using(self.session).get(**pk) # type: ignore[arg-type]
@cached_property
def kms_key(self) -> "KMSKey | None":
"""
Return the KMS key that is used to encrypt the data in the DB cluster.
.. note::
The output of this property is cached on the model instance, so
calling this multiple times will not result in multiple calls to the
AWS API. If you need a fresh copy of the data, you can re-get the
model instance from the manager.
"""
try:
pk = OrderedDict(
{
"KeyId": self.KmsKeyId,
}
)
except AttributeError:
return None
return KMSKey.objects.using(self.session).get(**pk) # type: ignore[arg-type]
[docs] def reboot(self, ForceFailover: "bool | None" = None) -> "DocDBInstance":
"""
You might need to reboot your instance, usually for maintenance reasons. For
example, if you make certain changes, or if you change the cluster parameter
group that is associated with the instance, you must reboot the instance for the
changes to take effect.
Keyword Args:
ForceFailover: When ``true``, the reboot is conducted through a Multi-AZ failover.
"""
return (
cast("DocDBInstanceManager", self.objects) # type: ignore[attr-defined]
.using(self.session)
.reboot(self.DBInstanceIdentifier, ForceFailover=ForceFailover)
)
[docs]class DocDBSubnetGroup(PrimaryBoto3Model):
"""
Detailed information about a subnet group.
"""
manager_class: ClassVar[type[Boto3ModelManager]] = DocDBSubnetGroupManager
DBSubnetGroupName: str
"""
The name of the subnet group.
"""
DBSubnetGroupDescription: str
"""
Provides the description of the subnet group.
"""
Subnets: "builtins.list[Subnet]"
"""
Detailed information about one or more subnets within a subnet group.
"""
VpcId: str
"""
Provides the virtual private cloud (VPC) ID of the subnet group.
"""
SubnetGroupStatus: "str | None" = "Complete"
"""
Provides the status of the subnet group.
"""
DBSubnetGroupArn: str = Field(default=None, frozen=True)
"""
The Amazon Resource Name (ARN) for the DB subnet group.
"""
SupportedNetworkTypes: "builtins.list[str]" = Field(
default_factory=list, frozen=True
)
"""
The network type of the DB subnet group.
"""
@property
def pk(self) -> str | None:
"""
Return the primary key of the model. This is the value of the
:py:attr:`DBSubnetGroupArn` attribute.
Returns:
The primary key of the model instance.
"""
return self.DBSubnetGroupArn
@property
def arn(self) -> str | None:
"""
Return the ARN of the model. This is the value of the
:py:attr:`DBSubnetGroupArn` attribute.
Returns:
The ARN of the model instance.
"""
return self.DBSubnetGroupArn
@property
def name(self) -> str | None:
"""
Return the name of the model. This is the value of the
:py:attr:`DBSubnetGroupName` attribute.
Returns:
The name of the model instance.
"""
return self.DBSubnetGroupName
def __hash__(self) -> int:
"""
Return the hash of the model.
This is the value of the
:py:attr:`DBSubnetGroupArn` attribute.
"""
return hash(self.DBSubnetGroupArn)
# =======================
# Request/Response Models
# =======================
[docs]class ServerlessV2ScalingConfiguration(Boto3Model):
"""
Sets the scaling configuration of an Amazon DocumentDB Serverless cluster.
"""
MinCapacity: "float | None" = None
"""
The minimum number of Amazon DocumentDB capacity units (DCUs) for an instance in an
Amazon DocumentDB Serverless cluster.
You can specify DCU values in half-step increments, such as 8, 8.5, 9, and so on.
"""
MaxCapacity: "float | None" = None
"""
The maximum number of Amazon DocumentDB capacity units (DCUs) for an instance in an
Amazon DocumentDB Serverless cluster.
You can specify DCU values in half-step increments, such as 32, 32.5, 33, and so on.
"""
[docs]class CreateDBClusterResult(Boto3Model):
DBCluster: "DocDBCluster | None" = None
"""
Detailed information about a cluster.
"""
[docs]class CloudwatchLogsExportConfiguration(Boto3Model):
"""
The configuration setting for the log types to be enabled for export to Amazon
CloudWatch Logs for a specific instance or cluster.
The ``EnableLogTypes`` and ``DisableLogTypes`` arrays determine which logs are exported (or not exported) to CloudWatch
Logs. The values within these arrays depend on the engine that is being used.
"""
EnableLogTypes: "builtins.list[str] | None" = Field(default_factory=list)
"""
The list of log types to enable.
"""
DisableLogTypes: "builtins.list[str] | None" = Field(default_factory=list)
"""
The list of log types to disable.
"""
[docs]class ModifyDBClusterResult(Boto3Model):
DBCluster: "DocDBCluster | None" = None
"""
Detailed information about a cluster.
"""
[docs]class DeleteDBClusterResult(Boto3Model):
DBCluster: "DocDBCluster | None" = None
"""
Detailed information about a cluster.
"""
[docs]class DocDBClusterMessage(Boto3Model):
"""
Represents the output of DescribeDBClusters.
"""
Marker: "str | None" = None
"""
An optional pagination token provided by a previous request.
If this parameter is specified, the response includes only
records beyond the marker, up to the value specified by ``MaxRecords``.
"""
DBClusters: "builtins.list[DocDBCluster] | None" = Field(default_factory=list)
"""
A list of clusters.
"""
[docs]class FailoverDBClusterResult(Boto3Model):
DBCluster: "DocDBCluster | None" = None
"""
Detailed information about a cluster.
"""
[docs]class RestoreDBClusterToPointInTimeResult(Boto3Model):
DBCluster: "DocDBCluster | None" = None
"""
Detailed information about a cluster.
"""
[docs]class StartDBClusterResult(Boto3Model):
DBCluster: "DocDBCluster | None" = None
"""
Detailed information about a cluster.
"""
[docs]class StopDBClusterResult(Boto3Model):
DBCluster: "DocDBCluster | None" = None
"""
Detailed information about a cluster.
"""
[docs]class DocDBCreateDBInstanceResult(Boto3Model):
DBInstance: "DocDBInstance | None" = None
"""
Detailed information about an instance.
"""
[docs]class DocDBModifyDBInstanceResult(Boto3Model):
DBInstance: "DocDBInstance | None" = None
"""
Detailed information about an instance.
"""
[docs]class DocDBDeleteDBInstanceResult(Boto3Model):
DBInstance: "DocDBInstance | None" = None
"""
Detailed information about an instance.
"""
[docs]class DocDBInstanceMessage(Boto3Model):
"""
Represents the output of DescribeDBInstances.
"""
Marker: "str | None" = None
"""
An optional pagination token provided by a previous request.
If this parameter is specified, the response includes only
records beyond the marker, up to the value specified by ``MaxRecords``.
"""
DBInstances: "builtins.list[DocDBInstance] | None" = Field(default_factory=list)
"""
Detailed information about one or more instances.
"""
[docs]class RebootDBInstanceResult(Boto3Model):
DBInstance: "DocDBInstance | None" = None
"""
Detailed information about an instance.
"""
[docs]class DocDBCreateDBSubnetGroupResult(Boto3Model):
DBSubnetGroup: "DocDBSubnetGroup | None" = None
"""
Detailed information about a subnet group.
"""
[docs]class DocDBModifyDBSubnetGroupResult(Boto3Model):
DBSubnetGroup: "DocDBSubnetGroup | None" = None
"""
Detailed information about a subnet group.
"""
[docs]class DocDBSubnetGroupMessage(Boto3Model):
"""
Represents the output of DescribeDBSubnetGroups.
"""
Marker: "str | None" = None
"""
An optional pagination token provided by a previous request.
If this parameter is specified, the response includes only
records beyond the marker, up to the value specified by ``MaxRecords``.
"""
DBSubnetGroups: "builtins.list[DocDBSubnetGroup] | None" = Field(
default_factory=list
)
"""
Detailed information about one or more subnet groups.
"""