# This file is automatically generated by botocraft. Do not edit directly.
# mypy: disable-error-code="index, override, assignment, union-attr, misc"
from botocraft.services.ec2 import Subnet, SubnetManager
from botocraft.mixins.elasticache import elasticache_user_group_add_tags
from .abstract import (
Boto3Model,
ReadonlyBoto3Model,
PrimaryBoto3Model,
ReadonlyPrimaryBoto3Model,
Boto3ModelManager,
ReadonlyBoto3ModelManager,
)
from botocraft.mixins.elasticache import elasticache_users_get_add_tags
from botocraft.mixins.elasticache import CacheClusterModelMixin
from .abstract import PrimaryBoto3ModelQuerySet
from datetime import datetime
from botocraft.mixins.elasticache import elasticache_user_add_tags
import builtins
from botocraft.services.common import Filter
from botocraft.mixins.elasticache import ElastiCacheManagerTagsMixin
from botocraft.services.ec2 import SecurityGroup, SecurityGroupManager
from functools import cached_property
from pydantic import Field
from botocraft.mixins.elasticache import elasticache_user_groups_get_add_tags
from collections import OrderedDict
from typing import ClassVar, Literal, Any, cast
from botocraft.mixins.tags import TagsDictMixin
from botocraft.mixins.elasticache import ReplicationGroupModelMixin
from botocraft.services.common import Tag
# ===============
# Managers
# ===============
[docs]class CacheClusterManager(ElastiCacheManagerTagsMixin, Boto3ModelManager):
service_name: str = "elasticache"
[docs] def create(
self,
model: "CacheCluster",
AZMode: "Literal['single-az', 'cross-az'] | None" = None,
PreferredAvailabilityZones: "builtins.list[str] | None" = None,
CacheParameterGroupName: "str | None" = None,
CacheSecurityGroupNames: "builtins.list[str] | None" = None,
SecurityGroupIds: "builtins.list[str] | None" = None,
Tags: "builtins.list[Tag] | None" = None,
SnapshotArns: "builtins.list[str] | None" = None,
SnapshotName: "str | None" = None,
Port: "int | None" = None,
NotificationTopicArn: "str | None" = None,
AuthToken: "str | None" = None,
OutpostMode: "Literal['single-outpost', 'cross-outpost'] | None" = None,
PreferredOutpostArns: "builtins.list[str] | None" = None,
) -> "CacheCluster":
"""
Creates a cluster. All nodes in the cluster run the same protocol-compliant
cache engine software, either Memcached, Valkey or Redis OSS.
Args:
model: The :py:class:`CacheCluster` to create.
Keyword Args:
AZMode: Specifies whether the nodes in this Memcached cluster are created in a single Availability Zone or created
across multiple Availability Zones in the cluster's region.
PreferredAvailabilityZones: A list of the Availability Zones in which cache nodes are created. The order of the
zones in the list is not important.
CacheParameterGroupName: The name of the parameter group to associate with this cluster. If this argument is
omitted, the default parameter group for the specified engine is used. You cannot use any parameter group which has
``cluster-enabled='yes'`` when creating a cluster.
CacheSecurityGroupNames: A list of security group names to associate with this cluster.
SecurityGroupIds: One or more VPC security groups associated with the cluster.
Tags: A list of tags to be added to this resource.
SnapshotArns: A single-element string list containing an Amazon Resource Name (ARN) that uniquely identifies a
Valkey or Redis OSS RDB snapshot file stored in Amazon S3. The snapshot file is used to populate the node group
(shard). The Amazon S3 object name in the ARN cannot contain any commas.
SnapshotName: The name of a Valkey or Redis OSS snapshot from which to restore data into the new node group (shard).
The snapshot status changes to ``restoring`` while the new node group (shard) is being created.
Port: The port number on which each of the cache nodes accepts connections.
NotificationTopicArn: The Amazon Resource Name (ARN) of the Amazon Simple Notification Service (SNS) topic to which
notifications are sent.
AuthToken: **Reserved parameter.** The password used to access a password protected server.
OutpostMode: Specifies whether the nodes in the cluster are created in a single outpost or across multiple outposts.
PreferredOutpostArns: The outpost ARNs in which the cache cluster is created.
"""
data = model.model_dump(exclude_none=True, by_alias=True)
args = dict(
CacheClusterId=data.get("CacheClusterId"),
ReplicationGroupId=data.get("ReplicationGroupId"),
AZMode=self.serialize(AZMode),
PreferredAvailabilityZone=data.get("PreferredAvailabilityZone"),
PreferredAvailabilityZones=self.serialize(PreferredAvailabilityZones),
NumCacheNodes=data.get("NumCacheNodes"),
CacheNodeType=data.get("CacheNodeType"),
Engine=data.get("Engine"),
EngineVersion=data.get("EngineVersion"),
CacheParameterGroupName=self.serialize(CacheParameterGroupName),
CacheSubnetGroupName=data.get("CacheSubnetGroupName"),
CacheSecurityGroupNames=self.serialize(CacheSecurityGroupNames),
SecurityGroupIds=self.serialize(SecurityGroupIds),
Tags=self.serialize(Tags),
SnapshotArns=self.serialize(SnapshotArns),
SnapshotName=self.serialize(SnapshotName),
PreferredMaintenanceWindow=data.get("PreferredMaintenanceWindow"),
Port=self.serialize(Port),
NotificationTopicArn=self.serialize(NotificationTopicArn),
AutoMinorVersionUpgrade=data.get("AutoMinorVersionUpgrade"),
SnapshotRetentionLimit=data.get("SnapshotRetentionLimit"),
SnapshotWindow=data.get("SnapshotWindow"),
AuthToken=self.serialize(AuthToken),
OutpostMode=self.serialize(OutpostMode),
PreferredOutpostArn=data.get("PreferredOutpostArn"),
PreferredOutpostArns=self.serialize(PreferredOutpostArns),
LogDeliveryConfigurations=data.get("LogDeliveryConfigurations"),
TransitEncryptionEnabled=data.get("TransitEncryptionEnabled"),
NetworkType=data.get("NetworkType"),
IpDiscovery=data.get("IpDiscovery"),
)
_response = self.client.create_cache_cluster(
**{k: v for k, v in args.items() if v is not None}
)
response = CreateCacheClusterResult(**_response)
self.sessionize(response.ElasticacheCluster)
return cast("CacheCluster", response.ElasticacheCluster)
[docs] def update(
self,
model: "CacheCluster",
CacheNodeIdsToRemove: "builtins.list[str] | None" = None,
AZMode: "Literal['single-az', 'cross-az'] | None" = None,
NewAvailabilityZones: "builtins.list[str] | None" = None,
CacheSecurityGroupNames: "builtins.list[str] | None" = None,
SecurityGroupIds: "builtins.list[str] | None" = None,
NotificationTopicArn: "str | None" = None,
CacheParameterGroupName: "str | None" = None,
NotificationTopicStatus: "str | None" = None,
ApplyImmediately: "bool | None" = None,
AuthToken: "str | None" = None,
AuthTokenUpdateStrategy: "Literal['SET', 'ROTATE', 'DELETE'] | None" = None,
ScaleConfig: "ElastiCacheScaleConfig | None" = None,
) -> "CacheCluster":
"""
Modifies the settings for a cluster. You can use this operation to change one or
more cluster configuration parameters by specifying the parameters and the new
values.
Args:
model: The :py:class:`CacheCluster` to update.
Keyword Args:
CacheNodeIdsToRemove: A list of cache node IDs to be removed. A node ID is a numeric identifier (0001, 0002, etc.).
This parameter is only valid when ``NumCacheNodes`` is less than the existing number of cache nodes. The number of
cache node IDs supplied in this parameter must match the difference between the existing number of cache nodes in
the cluster or pending cache nodes, whichever is greater, and the value of ``NumCacheNodes`` in the request.
AZMode: Specifies whether the new nodes in this Memcached cluster are all created in a single Availability Zone or
created across multiple Availability Zones.
NewAvailabilityZones: This option is only supported on Memcached clusters.
CacheSecurityGroupNames: A list of cache security group names to authorize on this cluster. This change is
asynchronously applied as soon as possible.
SecurityGroupIds: Specifies the VPC Security Groups associated with the cluster.
NotificationTopicArn: The Amazon Resource Name (ARN) of the Amazon SNS topic to which notifications are sent.
CacheParameterGroupName: The name of the cache parameter group to apply to this cluster. This change is
asynchronously applied as soon as possible for parameters when the ``ApplyImmediately`` parameter is specified as
``true`` for this request.
NotificationTopicStatus: The status of the Amazon SNS notification topic. Notifications are sent only if the status
is ``active``.
ApplyImmediately: If ``true``, this parameter causes the modifications in this request and any pending modifications
to be applied, asynchronously and as soon as possible, regardless of the ``PreferredMaintenanceWindow`` setting for
the cluster.
AuthToken: Reserved parameter. The password used to access a password protected server. This parameter must be
specified with the ``auth-token-update`` parameter. Password constraints:
AuthTokenUpdateStrategy: Specifies the strategy to use to update the AUTH token. This parameter must be specified
with the ``auth-token`` parameter. Possible values:
ScaleConfig: Configures horizontal or vertical scaling for Memcached clusters, specifying the scaling percentage and
interval.
"""
data = model.model_dump(exclude_none=True, by_alias=True)
args = dict(
CacheClusterId=data.get("CacheClusterId"),
NumCacheNodes=data.get("NumCacheNodes"),
CacheNodeIdsToRemove=self.serialize(CacheNodeIdsToRemove),
AZMode=self.serialize(AZMode),
NewAvailabilityZones=self.serialize(NewAvailabilityZones),
CacheSecurityGroupNames=self.serialize(CacheSecurityGroupNames),
SecurityGroupIds=self.serialize(SecurityGroupIds),
PreferredMaintenanceWindow=data.get("PreferredMaintenanceWindow"),
NotificationTopicArn=self.serialize(NotificationTopicArn),
CacheParameterGroupName=self.serialize(CacheParameterGroupName),
NotificationTopicStatus=self.serialize(NotificationTopicStatus),
ApplyImmediately=self.serialize(ApplyImmediately),
Engine=data.get("Engine"),
EngineVersion=data.get("EngineVersion"),
AutoMinorVersionUpgrade=data.get("AutoMinorVersionUpgrade"),
SnapshotRetentionLimit=data.get("SnapshotRetentionLimit"),
SnapshotWindow=data.get("SnapshotWindow"),
CacheNodeType=data.get("CacheNodeType"),
AuthToken=self.serialize(AuthToken),
AuthTokenUpdateStrategy=self.serialize(AuthTokenUpdateStrategy),
LogDeliveryConfigurations=data.get("LogDeliveryConfigurations"),
IpDiscovery=data.get("IpDiscovery"),
ScaleConfig=self.serialize(ScaleConfig),
)
_response = self.client.modify_cache_cluster(
**{k: v for k, v in args.items() if v is not None}
)
response = ModifyCacheClusterResult(**_response)
self.sessionize(response.ElasticacheCluster)
return cast("CacheCluster", response.ElasticacheCluster)
[docs] def delete(
self, CacheClusterId: str, *, FinalSnapshotIdentifier: "str | None" = None
) -> "CacheCluster":
"""
Deletes a previously provisioned cluster. ``DeleteCacheCluster`` deletes all
associated cache nodes, node endpoints and the cluster itself. When you receive
a successful response from this operation, Amazon ElastiCache immediately begins
deleting the cluster; you cannot cancel or revert this operation.
Args:
CacheClusterId: The cluster identifier for the cluster to be deleted. This parameter is not case sensitive.
Keyword Args:
FinalSnapshotIdentifier: The user-supplied name of a final cluster snapshot. This is the unique name that identifies
the snapshot. ElastiCache creates the snapshot, and then deletes the cluster immediately afterward.
"""
args: dict[str, Any] = dict(
CacheClusterId=self.serialize(CacheClusterId),
FinalSnapshotIdentifier=self.serialize(FinalSnapshotIdentifier),
)
_response = self.client.delete_cache_cluster(
**{k: v for k, v in args.items() if v is not None}
)
response = DeleteCacheClusterResult(**_response)
return cast("CacheCluster", response.ElasticacheCluster)
[docs] def get(
self,
CacheClusterId: str,
*,
ShowCacheNodeInfo: bool = True,
ShowCacheClustersNotInReplicationGroups: "bool | None" = None,
) -> "CacheCluster | None":
"""
Returns information about all provisioned clusters if no cluster identifier is
specified, or about a specific cache cluster if a cluster identifier is
supplied.
Args:
CacheClusterId: The user-supplied cluster identifier. If this parameter is specified, only information about that
specific cluster is returned. This parameter isn't case sensitive.
Keyword Args:
ShowCacheNodeInfo: An optional flag that can be included in the ``DescribeCacheCluster`` request to retrieve
information about the individual cache nodes.
ShowCacheClustersNotInReplicationGroups: An optional flag that can be included in the ``DescribeCacheCluster``
request to show only nodes (API/CLI: clusters) that are not members of a replication group. In practice, this means
Memcached and single node Valkey or Redis OSS clusters.
"""
args: dict[str, Any] = dict(
CacheClusterId=self.serialize(CacheClusterId),
ShowCacheNodeInfo=self.serialize(ShowCacheNodeInfo),
ShowCacheClustersNotInReplicationGroups=self.serialize(
ShowCacheClustersNotInReplicationGroups
),
)
_response = self.client.describe_cache_clusters(
**{k: v for k, v in args.items() if v is not None}
)
response = CacheClusterMessage(**_response)
if response and response.CacheClusters:
self.sessionize(response.CacheClusters[0])
return response.CacheClusters[0]
return None
[docs] def list(
self,
*,
CacheClusterId: "str | None" = None,
ShowCacheNodeInfo: bool = True,
ShowCacheClustersNotInReplicationGroups: "bool | None" = None,
) -> PrimaryBoto3ModelQuerySet:
"""
Returns information about all provisioned clusters if no cluster identifier is
specified, or about a specific cache cluster if a cluster identifier is
supplied.
Keyword Args:
CacheClusterId: The user-supplied cluster identifier. If this parameter is specified, only information about that
specific cluster is returned. This parameter isn't case sensitive.
ShowCacheNodeInfo: An optional flag that can be included in the ``DescribeCacheCluster`` request to retrieve
information about the individual cache nodes.
ShowCacheClustersNotInReplicationGroups: An optional flag that can be included in the ``DescribeCacheCluster``
request to show only nodes (API/CLI: clusters) that are not members of a replication group. In practice, this means
Memcached and single node Valkey or Redis OSS clusters.
"""
paginator = self.client.get_paginator("describe_cache_clusters")
args: dict[str, Any] = dict(
CacheClusterId=self.serialize(CacheClusterId),
ShowCacheNodeInfo=self.serialize(ShowCacheNodeInfo),
ShowCacheClustersNotInReplicationGroups=self.serialize(
ShowCacheClustersNotInReplicationGroups
),
)
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 = CacheClusterMessage(**_response)
if response.CacheClusters:
results.extend(response.CacheClusters)
else:
if getattr(response, "NextToken", None):
continue
break
self.sessionize(results)
if results and isinstance(results[0], Boto3Model):
return PrimaryBoto3ModelQuerySet(results)
return results
[docs]class CacheParameterGroupManager(Boto3ModelManager):
service_name: str = "elasticache"
[docs] def create(
self, model: "CacheParameterGroup", Tags: "builtins.list[Tag] | None" = None
) -> "CacheParameterGroup":
"""
Creates a new Amazon ElastiCache cache parameter group. An ElastiCache cache
parameter group is a collection of parameters and their values that are applied
to all of the nodes in any cluster or replication group using the
CacheParameterGroup.
Args:
model: The :py:class:`CacheParameterGroup` to create.
Keyword Args:
Tags: A list of tags to be added to this resource. A tag is a key-value pair. A tag key must be accompanied by a tag
value, although null is accepted.
"""
data = model.model_dump(exclude_none=True, by_alias=True)
args = dict(
CacheParameterGroupName=data.get("CacheParameterGroupName"),
CacheParameterGroupFamily=data.get("CacheParameterGroupFamily"),
Description=data.get("Description"),
Tags=self.serialize(Tags),
)
_response = self.client.create_cache_parameter_group(
**{k: v for k, v in args.items() if v is not None}
)
response = CreateCacheParameterGroupResult(**_response)
self.sessionize(response.ElasticacheParameterGroup)
return cast("CacheParameterGroup", response.ElasticacheParameterGroup)
[docs] def update(self, model: "CacheParameterGroup") -> str:
"""
Modifies the parameters of a cache parameter group. You can modify up to 20
parameters in a single request by submitting a list parameter name and value
pairs.
Args:
model: The :py:class:`CacheParameterGroup` to update.
"""
data = model.model_dump(exclude_none=True, by_alias=True)
args = dict(
CacheParameterGroupName=data.get("CacheParameterGroupName"),
ParameterNameValues=data.get("ParameterNameValues"),
)
_response = self.client.modify_cache_parameter_group(
**{k: v for k, v in args.items() if v is not None}
)
response = CacheParameterGroupNameMessage(**_response)
self.sessionize(response.CacheParameterGroupName)
return cast("str", response.CacheParameterGroupName)
[docs] def delete(self, CacheParameterGroupName: str) -> None:
"""
Deletes the specified cache parameter group. You cannot delete a cache parameter
group if it is associated with any cache clusters. You cannot delete the default
cache parameter groups in your account.
Args:
CacheParameterGroupName: The name of the cache parameter group to delete.
"""
args: dict[str, Any] = dict(
CacheParameterGroupName=self.serialize(CacheParameterGroupName)
)
self.client.delete_cache_parameter_group(
**{k: v for k, v in args.items() if v is not None}
)
[docs] def get(self, CacheParameterGroupName: str) -> "CacheParameterGroup | None":
"""
Returns a list of cache parameter group descriptions. If a cache parameter group
name is specified, the list contains only the descriptions for that group.
Args:
CacheParameterGroupName: The name of a specific cache parameter group to return details for.
"""
args: dict[str, Any] = dict(
CacheParameterGroupName=self.serialize(CacheParameterGroupName)
)
_response = self.client.describe_cache_parameter_groups(
**{k: v for k, v in args.items() if v is not None}
)
response = CacheParameterGroupsMessage(**_response)
if response and response.CacheParameterGroups:
self.sessionize(response.CacheParameterGroups[0])
return response.CacheParameterGroups[0]
return None
[docs] def list(
self, *, CacheParameterGroupName: "str | None" = None
) -> PrimaryBoto3ModelQuerySet:
"""
Returns a list of cache parameter group descriptions. If a cache parameter group
name is specified, the list contains only the descriptions for that group.
Keyword Args:
CacheParameterGroupName: The name of a specific cache parameter group to return details for.
"""
paginator = self.client.get_paginator("describe_cache_parameter_groups")
args: dict[str, Any] = dict(
CacheParameterGroupName=self.serialize(CacheParameterGroupName)
)
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 = CacheParameterGroupsMessage(**_response)
if response.CacheParameterGroups:
results.extend(response.CacheParameterGroups)
else:
if getattr(response, "NextToken", None):
continue
break
self.sessionize(results)
if results and isinstance(results[0], Boto3Model):
return PrimaryBoto3ModelQuerySet(results)
return results
[docs] def reset(
self,
CacheParameterGroupName: str,
*,
ResetAllParameters: "bool | None" = None,
ParameterNameValues: "builtins.list[ParameterNameValue] | None" = None,
) -> str:
"""
Modifies the parameters of a cache parameter group to the engine or system
default value. You can reset specific parameters by submitting a list of
parameter names. To reset the entire cache parameter group, specify the
``ResetAllParameters`` and ``CacheParameterGroupName`` parameters.
Args:
CacheParameterGroupName: The name of the cache parameter group to reset.
Keyword Args:
ResetAllParameters: If ``true``, all parameters in the cache parameter group are reset to their default values. If
``false``, only the parameters listed by ``ParameterNameValues`` are reset to their default values.
ParameterNameValues: An array of parameter names to reset to their default values. If ``ResetAllParameters`` is
``true``, do not use ``ParameterNameValues``. If ``ResetAllParameters`` is ``false``, you must specify the name of
at least one parameter to reset.
"""
args: dict[str, Any] = dict(
CacheParameterGroupName=self.serialize(CacheParameterGroupName),
ResetAllParameters=self.serialize(ResetAllParameters),
ParameterNameValues=self.serialize(ParameterNameValues),
)
_response = self.client.reset_cache_parameter_group(
**{k: v for k, v in args.items() if v is not None}
)
response = CacheParameterGroupNameMessage(**_response)
results: str = None
if response is not None:
results = response.CacheParameterGroupName
self.sessionize(results)
return cast("str", results)
[docs] def parameters(
self, CacheParameterGroupName: str, *, Source: "str | None" = None
) -> "builtins.list[CacheParameter]":
"""
Returns the detailed parameter list for a particular cache parameter group.
Args:
CacheParameterGroupName: The name of a specific cache parameter group to return details for.
Keyword Args:
Source: The parameter types to return.
"""
paginator = self.client.get_paginator("describe_cache_parameters")
args: dict[str, Any] = dict(
CacheParameterGroupName=self.serialize(CacheParameterGroupName),
Source=self.serialize(Source),
)
response_iterator = paginator.paginate(
**{k: v for k, v in args.items() if v is not None}
)
results: "builtins.list[CacheParameter]" = []
for _response in response_iterator:
response = CacheParameterGroupDetails(**_response)
if response.Parameters is not None:
results.extend(response.Parameters)
else:
break
self.sessionize(results)
return cast("builtins.list[CacheParameter]", results)
[docs]class CacheSubnetGroupManager(Boto3ModelManager):
service_name: str = "elasticache"
[docs] def create(
self, model: "CacheSubnetGroup", Tags: "builtins.list[Tag] | None" = None
) -> "CacheSubnetGroup":
"""
Creates a new cache subnet group.
Args:
model: The :py:class:`CacheSubnetGroup` to create.
Keyword Args:
Tags: A list of tags to be added to this resource. A tag is a key-value pair. A tag key must be accompanied by a tag
value, although null is accepted.
"""
data = model.model_dump(exclude_none=True, by_alias=True)
args = dict(
CacheSubnetGroupName=data.get("CacheSubnetGroupName"),
CacheSubnetGroupDescription=data.get("CacheSubnetGroupDescription"),
SubnetIds=data.get("SubnetIds"),
Tags=self.serialize(Tags),
)
_response = self.client.create_cache_subnet_group(
**{k: v for k, v in args.items() if v is not None}
)
response = CreateCacheSubnetGroupResult(**_response)
self.sessionize(response.ElasticacheSubnetGroup)
return cast("CacheSubnetGroup", response.ElasticacheSubnetGroup)
[docs] def update(
self, model: "CacheSubnetGroup", SubnetIds: "builtins.list[str] | None" = None
) -> "CacheSubnetGroup":
"""
Modifies an existing cache subnet group.
Args:
model: The :py:class:`CacheSubnetGroup` to update.
Keyword Args:
SubnetIds: The EC2 subnet IDs for the cache subnet group.
"""
data = model.model_dump(exclude_none=True, by_alias=True)
args = dict(
CacheSubnetGroupName=data.get("CacheSubnetGroupName"),
CacheSubnetGroupDescription=data.get("CacheSubnetGroupDescription"),
SubnetIds=self.serialize(SubnetIds),
)
_response = self.client.modify_cache_subnet_group(
**{k: v for k, v in args.items() if v is not None}
)
response = ModifyCacheSubnetGroupResult(**_response)
self.sessionize(response.ElasticacheSubnetGroup)
return cast("CacheSubnetGroup", response.ElasticacheSubnetGroup)
[docs] def delete(self, CacheSubnetGroupName: str) -> None:
"""
Deletes a cache subnet group.
Args:
CacheSubnetGroupName: The name of the cache subnet group to delete.
"""
args: dict[str, Any] = dict(
CacheSubnetGroupName=self.serialize(CacheSubnetGroupName)
)
self.client.delete_cache_subnet_group(
**{k: v for k, v in args.items() if v is not None}
)
[docs] def get(self, CacheSubnetGroupName: str) -> "CacheSubnetGroup | None":
"""
Returns a list of cache subnet group descriptions. If a subnet group name is
specified, the list contains only the description of that group. This is
applicable only when you have ElastiCache in VPC setup. All ElastiCache clusters
now launch in VPC by default.
Args:
CacheSubnetGroupName: The name of the cache subnet group to return details for.
"""
args: dict[str, Any] = dict(
CacheSubnetGroupName=self.serialize(CacheSubnetGroupName)
)
_response = self.client.describe_cache_subnet_groups(
**{k: v for k, v in args.items() if v is not None}
)
response = CacheSubnetGroupMessage(**_response)
if response and response.CacheSubnetGroups:
self.sessionize(response.CacheSubnetGroups[0])
return response.CacheSubnetGroups[0]
return None
[docs] def list(
self, *, CacheSubnetGroupName: "str | None" = None
) -> PrimaryBoto3ModelQuerySet:
"""
Returns a list of cache subnet group descriptions. If a subnet group name is
specified, the list contains only the description of that group. This is
applicable only when you have ElastiCache in VPC setup. All ElastiCache clusters
now launch in VPC by default.
Keyword Args:
CacheSubnetGroupName: The name of the cache subnet group to return details for.
"""
paginator = self.client.get_paginator("describe_cache_subnet_groups")
args: dict[str, Any] = dict(
CacheSubnetGroupName=self.serialize(CacheSubnetGroupName)
)
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 = CacheSubnetGroupMessage(**_response)
if response.CacheSubnetGroups:
results.extend(response.CacheSubnetGroups)
else:
if getattr(response, "NextToken", None):
continue
break
self.sessionize(results)
if results and isinstance(results[0], Boto3Model):
return PrimaryBoto3ModelQuerySet(results)
return results
[docs]class CacheSecurityGroupManager(Boto3ModelManager):
service_name: str = "elasticache"
[docs] def create(
self, model: "CacheSecurityGroup", Tags: "builtins.list[Tag] | None" = None
) -> "CacheSecurityGroup":
"""
Creates a new cache security group. Use a cache security group to control access
to one or more clusters.
Args:
model: The :py:class:`CacheSecurityGroup` to create.
Keyword Args:
Tags: A list of tags to be added to this resource. A tag is a key-value pair. A tag key must be accompanied by a tag
value, although null is accepted.
"""
data = model.model_dump(exclude_none=True, by_alias=True)
args = dict(
CacheSecurityGroupName=data.get("CacheSecurityGroupName"),
Description=data.get("Description"),
Tags=self.serialize(Tags),
)
_response = self.client.create_cache_security_group(
**{k: v for k, v in args.items() if v is not None}
)
response = CreateCacheSecurityGroupResult(**_response)
self.sessionize(response.ElasticacheSecurityGroup)
return cast("CacheSecurityGroup", response.ElasticacheSecurityGroup)
[docs] def delete(self, CacheSecurityGroupName: str) -> None:
"""
Deletes a cache security group.
Args:
CacheSecurityGroupName: The name of the cache security group to delete.
"""
args: dict[str, Any] = dict(
CacheSecurityGroupName=self.serialize(CacheSecurityGroupName)
)
self.client.delete_cache_security_group(
**{k: v for k, v in args.items() if v is not None}
)
[docs] def get(self, CacheSecurityGroupName: str) -> "CacheSecurityGroup | None":
"""
Returns a list of cache security group descriptions. If a cache security group
name is specified, the list contains only the description of that group. This
applicable only when you have ElastiCache in Classic setup.
Args:
CacheSecurityGroupName: The name of the cache security group to return details for.
"""
args: dict[str, Any] = dict(
CacheSecurityGroupName=self.serialize(CacheSecurityGroupName)
)
_response = self.client.describe_cache_security_groups(
**{k: v for k, v in args.items() if v is not None}
)
response = CacheSecurityGroupMessage(**_response)
if response and response.CacheSecurityGroups:
self.sessionize(response.CacheSecurityGroups[0])
return response.CacheSecurityGroups[0]
return None
[docs] def list(
self, *, CacheSecurityGroupName: "str | None" = None
) -> PrimaryBoto3ModelQuerySet:
"""
Returns a list of cache security group descriptions. If a cache security group
name is specified, the list contains only the description of that group. This
applicable only when you have ElastiCache in Classic setup.
Keyword Args:
CacheSecurityGroupName: The name of the cache security group to return details for.
"""
paginator = self.client.get_paginator("describe_cache_security_groups")
args: dict[str, Any] = dict(
CacheSecurityGroupName=self.serialize(CacheSecurityGroupName)
)
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 = CacheSecurityGroupMessage(**_response)
if response.CacheSecurityGroups:
results.extend(response.CacheSecurityGroups)
else:
if getattr(response, "NextToken", None):
continue
break
self.sessionize(results)
if results and isinstance(results[0], Boto3Model):
return PrimaryBoto3ModelQuerySet(results)
return results
[docs] def authorize_ingress(
self,
CacheSecurityGroupName: str,
EC2SecurityGroupName: str,
EC2SecurityGroupOwnerId: str,
) -> "CacheSecurityGroup":
"""
Allows network ingress to a cache security group. Applications using ElastiCache
must be running on Amazon EC2, and Amazon EC2 security groups are used as the
authorization mechanism.
Args:
CacheSecurityGroupName: The cache security group that allows network ingress.
EC2SecurityGroupName: The Amazon EC2 security group to be authorized for ingress to the cache security group.
EC2SecurityGroupOwnerId: The Amazon account number of the Amazon EC2 security group owner. Note that this is not the
same thing as an Amazon access key ID - you must provide a valid Amazon account number for this parameter.
"""
args: dict[str, Any] = dict(
CacheSecurityGroupName=self.serialize(CacheSecurityGroupName),
EC2SecurityGroupName=self.serialize(EC2SecurityGroupName),
EC2SecurityGroupOwnerId=self.serialize(EC2SecurityGroupOwnerId),
)
_response = self.client.authorize_cache_security_group_ingress(
**{k: v for k, v in args.items() if v is not None}
)
response = AuthorizeCacheSecurityGroupIngressResult(**_response)
results: "CacheSecurityGroup" = None
if response is not None:
results = response.ElasticacheSecurityGroup
self.sessionize(results)
return cast("CacheSecurityGroup", results)
[docs] def revoke_ingress(
self,
CacheSecurityGroupName: str,
EC2SecurityGroupName: str,
EC2SecurityGroupOwnerId: str,
) -> "CacheSecurityGroup":
"""
Revokes ingress from a cache security group. Use this operation to disallow
access from an Amazon EC2 security group that had been previously authorized.
Args:
CacheSecurityGroupName: The name of the cache security group to revoke ingress from.
EC2SecurityGroupName: The name of the Amazon EC2 security group to revoke access from.
EC2SecurityGroupOwnerId: The Amazon account number of the Amazon EC2 security group owner. Note that this is not the
same thing as an Amazon access key ID - you must provide a valid Amazon account number for this parameter.
"""
args: dict[str, Any] = dict(
CacheSecurityGroupName=self.serialize(CacheSecurityGroupName),
EC2SecurityGroupName=self.serialize(EC2SecurityGroupName),
EC2SecurityGroupOwnerId=self.serialize(EC2SecurityGroupOwnerId),
)
_response = self.client.revoke_cache_security_group_ingress(
**{k: v for k, v in args.items() if v is not None}
)
response = RevokeCacheSecurityGroupIngressResult(**_response)
results: "CacheSecurityGroup" = None
if response is not None:
results = response.ElasticacheSecurityGroup
self.sessionize(results)
return cast("CacheSecurityGroup", results)
[docs]class ReplicationGroupManager(ElastiCacheManagerTagsMixin, Boto3ModelManager):
service_name: str = "elasticache"
[docs] def create(
self,
model: "ReplicationGroup",
ReplicationGroupDescription: str,
GlobalReplicationGroupId: "str | None" = None,
PrimaryClusterId: "str | None" = None,
AutomaticFailoverEnabled: "bool | None" = None,
MultiAZEnabled: "bool | None" = None,
NumCacheClusters: "int | None" = None,
PreferredCacheClusterAZs: "builtins.list[str] | None" = None,
NumNodeGroups: "int | None" = None,
ReplicasPerNodeGroup: "int | None" = None,
NodeGroupConfiguration: "builtins.list[NodeGroupConfiguration] | None" = None,
EngineVersion: "str | None" = None,
CacheParameterGroupName: "str | None" = None,
CacheSubnetGroupName: "str | None" = None,
CacheSecurityGroupNames: "builtins.list[str] | None" = None,
SecurityGroupIds: "builtins.list[str] | None" = None,
Tags: "builtins.list[Tag] | None" = None,
SnapshotArns: "builtins.list[str] | None" = None,
SnapshotName: "str | None" = None,
PreferredMaintenanceWindow: "str | None" = None,
Port: "int | None" = None,
NotificationTopicArn: "str | None" = None,
AuthToken: "str | None" = None,
DataTieringEnabled: "bool | None" = None,
ServerlessCacheSnapshotName: "str | None" = None,
) -> "ReplicationGroup":
"""
Creates a Valkey or Redis OSS (cluster mode disabled) or a Valkey or Redis OSS
(cluster mode enabled) replication group.
Args:
model: The :py:class:`ReplicationGroup` to create.
ReplicationGroupDescription: A user-created description for the replication group.
Keyword Args:
GlobalReplicationGroupId: The name of the Global datastore
PrimaryClusterId: The identifier of the cluster that serves as the primary for this replication group. This cluster
must already exist and have a status of ``available``.
AutomaticFailoverEnabled: Specifies whether a read-only replica is automatically promoted to read/write primary if
the existing primary fails.
MultiAZEnabled: A flag indicating if you have Multi-AZ enabled to enhance fault tolerance. For more information, see
`Minimizing Downtime: Multi-AZ <http://docs.aws.amazon.com/AmazonElastiCache/latest/dg/AutoFailover.html>`_.
NumCacheClusters: The number of clusters this replication group initially has.
PreferredCacheClusterAZs: A list of EC2 Availability Zones in which the replication group's clusters are created.
The order of the Availability Zones in the list is the order in which clusters are allocated. The primary cluster is
created in the first AZ in the list.
NumNodeGroups: An optional parameter that specifies the number of node groups (shards) for this Valkey or Redis OSS
(cluster mode enabled) replication group. For Valkey or Redis OSS (cluster mode disabled) either omit this parameter
or set it to 1.
ReplicasPerNodeGroup: An optional parameter that specifies the number of replica nodes in each node group (shard).
Valid values are 0 to 5.
NodeGroupConfiguration: A list of node group (shard) configuration options. Each node group (shard) configuration
has the following members: ``PrimaryAvailabilityZone``, ``ReplicaAvailabilityZones``, ``ReplicaCount``, and
``Slots``.
EngineVersion: The version number of the cache engine to be used for the clusters in this replication group. To view
the supported cache engine versions, use the ``DescribeCacheEngineVersions`` operation.
CacheParameterGroupName: The name of the parameter group to associate with this replication group. If this argument
is omitted, the default cache parameter group for the specified engine is used.
CacheSubnetGroupName: The name of the cache subnet group to be used for the replication group.
CacheSecurityGroupNames: A list of cache security group names to associate with this replication group.
SecurityGroupIds: One or more Amazon VPC security groups associated with this replication group.
Tags: A list of tags to be added to this resource. Tags are comma-separated key,value pairs (e.g. Key=``myKey``,
Value=``myKeyValue``. You can include multiple tags as shown following: Key=``myKey``, Value=``myKeyValue``
Key=``mySecondKey``, Value=``mySecondKeyValue``. Tags on replication groups will be replicated to all nodes.
SnapshotArns: A list of Amazon Resource Names (ARN) that uniquely identify the Valkey or Redis OSS RDB snapshot
files stored in Amazon S3. The snapshot files are used to populate the new replication group. The Amazon S3 object
name in the ARN cannot contain any commas. The new replication group will have the number of node groups (console:
shards) specified by the parameter *NumNodeGroups* or the number of node groups configured by
*NodeGroupConfiguration* regardless of the number of ARNs specified here.
SnapshotName: The name of a snapshot from which to restore data into the new replication group. The snapshot status
changes to ``restoring`` while the new replication group is being created.
PreferredMaintenanceWindow: Specifies the weekly time range during which maintenance on the cluster is performed. It
is specified as a range in the format ddd:hh24:mi-ddd:hh24:mi (24H Clock UTC). The minimum maintenance window is a
60 minute period.
Port: The port number on which each member of the replication group accepts connections.
NotificationTopicArn: The Amazon Resource Name (ARN) of the Amazon Simple Notification Service (SNS) topic to which
notifications are sent.
AuthToken: **Reserved parameter.** The password used to access a password protected server.
DataTieringEnabled: Enables data tiering. Data tiering is only supported for replication groups using the r6gd node
type. This parameter must be set to true when using r6gd nodes. For more information, see `Data tiering
<https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/data-tiering.html>`_.
ServerlessCacheSnapshotName: The name of the snapshot used to create a replication group. Available for Valkey,
Redis OSS only.
"""
data = model.model_dump(exclude_none=True, by_alias=True)
args = dict(
ReplicationGroupId=data.get("ReplicationGroupId"),
ReplicationGroupDescription=self.serialize(ReplicationGroupDescription),
GlobalReplicationGroupId=self.serialize(GlobalReplicationGroupId),
PrimaryClusterId=self.serialize(PrimaryClusterId),
AutomaticFailoverEnabled=self.serialize(AutomaticFailoverEnabled),
MultiAZEnabled=self.serialize(MultiAZEnabled),
NumCacheClusters=self.serialize(NumCacheClusters),
PreferredCacheClusterAZs=self.serialize(PreferredCacheClusterAZs),
NumNodeGroups=self.serialize(NumNodeGroups),
ReplicasPerNodeGroup=self.serialize(ReplicasPerNodeGroup),
NodeGroupConfiguration=self.serialize(NodeGroupConfiguration),
CacheNodeType=data.get("CacheNodeType"),
Engine=data.get("Engine"),
EngineVersion=self.serialize(EngineVersion),
CacheParameterGroupName=self.serialize(CacheParameterGroupName),
CacheSubnetGroupName=self.serialize(CacheSubnetGroupName),
CacheSecurityGroupNames=self.serialize(CacheSecurityGroupNames),
SecurityGroupIds=self.serialize(SecurityGroupIds),
Tags=self.serialize(Tags),
SnapshotArns=self.serialize(SnapshotArns),
SnapshotName=self.serialize(SnapshotName),
PreferredMaintenanceWindow=self.serialize(PreferredMaintenanceWindow),
Port=self.serialize(Port),
NotificationTopicArn=self.serialize(NotificationTopicArn),
AutoMinorVersionUpgrade=data.get("AutoMinorVersionUpgrade"),
SnapshotRetentionLimit=data.get("SnapshotRetentionLimit"),
SnapshotWindow=data.get("SnapshotWindow"),
AuthToken=self.serialize(AuthToken),
TransitEncryptionEnabled=data.get("TransitEncryptionEnabled"),
AtRestEncryptionEnabled=data.get("AtRestEncryptionEnabled"),
KmsKeyId=data.get("KmsKeyId"),
UserGroupIds=data.get("UserGroupIds"),
LogDeliveryConfigurations=data.get("LogDeliveryConfigurations"),
DataTieringEnabled=self.serialize(DataTieringEnabled),
NetworkType=data.get("NetworkType"),
IpDiscovery=data.get("IpDiscovery"),
TransitEncryptionMode=data.get("TransitEncryptionMode"),
ClusterMode=data.get("ClusterMode"),
ServerlessCacheSnapshotName=self.serialize(ServerlessCacheSnapshotName),
)
_response = self.client.create_replication_group(
**{k: v for k, v in args.items() if v is not None}
)
response = CreateReplicationGroupResult(**_response)
self.sessionize(response.ElasticacheReplicationGroup)
return cast("ReplicationGroup", response.ElasticacheReplicationGroup)
[docs] def update(
self,
model: "ReplicationGroup",
ReplicationGroupDescription: "str | None" = None,
PrimaryClusterId: "str | None" = None,
AutomaticFailoverEnabled: "bool | None" = None,
MultiAZEnabled: "bool | None" = None,
NodeGroupId: "str | None" = None,
CacheSecurityGroupNames: "builtins.list[str] | None" = None,
SecurityGroupIds: "builtins.list[str] | None" = None,
PreferredMaintenanceWindow: "str | None" = None,
NotificationTopicArn: "str | None" = None,
CacheParameterGroupName: "str | None" = None,
NotificationTopicStatus: "str | None" = None,
ApplyImmediately: "bool | None" = None,
EngineVersion: "str | None" = None,
AuthToken: "str | None" = None,
AuthTokenUpdateStrategy: "Literal['SET', 'ROTATE', 'DELETE'] | None" = None,
UserGroupIdsToAdd: "builtins.list[str] | None" = None,
UserGroupIdsToRemove: "builtins.list[str] | None" = None,
RemoveUserGroups: "bool | None" = None,
) -> "ReplicationGroup":
"""
Modifies the settings for a replication group. This is limited to Valkey and
Redis OSS 7 and above.
Args:
model: The :py:class:`ReplicationGroup` to update.
Keyword Args:
ReplicationGroupDescription: A description for the replication group. Maximum length is 255 characters.
PrimaryClusterId: For replication groups with a single primary, if this parameter is specified, ElastiCache promotes
the specified cluster in the specified replication group to the primary role. The nodes of all other clusters in the
replication group are read replicas.
AutomaticFailoverEnabled: Determines whether a read replica is automatically promoted to read/write primary if the
existing primary encounters a failure.
MultiAZEnabled: A flag to indicate MultiAZ is enabled.
NodeGroupId: Deprecated. This parameter is not used.
CacheSecurityGroupNames: A list of cache security group names to authorize for the clusters in this replication
group. This change is asynchronously applied as soon as possible.
SecurityGroupIds: Specifies the VPC Security Groups associated with the clusters in the replication group.
PreferredMaintenanceWindow: Specifies the weekly time range during which maintenance on the cluster is performed. It
is specified as a range in the format ddd:hh24:mi-ddd:hh24:mi (24H Clock UTC). The minimum maintenance window is a
60 minute period.
NotificationTopicArn: The Amazon Resource Name (ARN) of the Amazon SNS topic to which notifications are sent.
CacheParameterGroupName: The name of the cache parameter group to apply to all of the clusters in this replication
group. This change is asynchronously applied as soon as possible for parameters when the ``ApplyImmediately``
parameter is specified as ``true`` for this request.
NotificationTopicStatus: The status of the Amazon SNS notification topic for the replication group. Notifications
are sent only if the status is ``active``.
ApplyImmediately: If ``true``, this parameter causes the modifications in this request and any pending modifications
to be applied, asynchronously and as soon as possible, regardless of the ``PreferredMaintenanceWindow`` setting for
the replication group.
EngineVersion: The upgraded version of the cache engine to be run on the clusters in the replication group.
AuthToken: Reserved parameter. The password used to access a password protected server. This parameter must be
specified with the ``auth-token-update-strategy`` parameter. Password constraints:
AuthTokenUpdateStrategy: Specifies the strategy to use to update the AUTH token. This parameter must be specified
with the ``auth-token`` parameter. Possible values:
UserGroupIdsToAdd: The ID of the user group you are associating with the replication group.
UserGroupIdsToRemove: The ID of the user group to disassociate from the replication group, meaning the users in the
group no longer can access the replication group.
RemoveUserGroups: Removes the user group associated with this replication group.
"""
data = model.model_dump(exclude_none=True, by_alias=True)
args = dict(
ReplicationGroupId=data.get("ReplicationGroupId"),
ReplicationGroupDescription=self.serialize(ReplicationGroupDescription),
PrimaryClusterId=self.serialize(PrimaryClusterId),
SnapshottingClusterId=data.get("SnapshottingClusterId"),
AutomaticFailoverEnabled=self.serialize(AutomaticFailoverEnabled),
MultiAZEnabled=self.serialize(MultiAZEnabled),
NodeGroupId=self.serialize(NodeGroupId),
CacheSecurityGroupNames=self.serialize(CacheSecurityGroupNames),
SecurityGroupIds=self.serialize(SecurityGroupIds),
PreferredMaintenanceWindow=self.serialize(PreferredMaintenanceWindow),
NotificationTopicArn=self.serialize(NotificationTopicArn),
CacheParameterGroupName=self.serialize(CacheParameterGroupName),
NotificationTopicStatus=self.serialize(NotificationTopicStatus),
ApplyImmediately=self.serialize(ApplyImmediately),
Engine=data.get("Engine"),
EngineVersion=self.serialize(EngineVersion),
AutoMinorVersionUpgrade=data.get("AutoMinorVersionUpgrade"),
SnapshotRetentionLimit=data.get("SnapshotRetentionLimit"),
SnapshotWindow=data.get("SnapshotWindow"),
CacheNodeType=data.get("CacheNodeType"),
AuthToken=self.serialize(AuthToken),
AuthTokenUpdateStrategy=self.serialize(AuthTokenUpdateStrategy),
UserGroupIdsToAdd=self.serialize(UserGroupIdsToAdd),
UserGroupIdsToRemove=self.serialize(UserGroupIdsToRemove),
RemoveUserGroups=self.serialize(RemoveUserGroups),
LogDeliveryConfigurations=data.get("LogDeliveryConfigurations"),
IpDiscovery=data.get("IpDiscovery"),
TransitEncryptionEnabled=data.get("TransitEncryptionEnabled"),
TransitEncryptionMode=data.get("TransitEncryptionMode"),
ClusterMode=data.get("ClusterMode"),
)
_response = self.client.modify_replication_group(
**{k: v for k, v in args.items() if v is not None}
)
response = ModifyReplicationGroupResult(**_response)
self.sessionize(response.ElasticacheReplicationGroup)
return cast("ReplicationGroup", response.ElasticacheReplicationGroup)
[docs] def delete(
self,
ReplicationGroupId: str,
*,
RetainPrimaryCluster: "bool | None" = None,
FinalSnapshotIdentifier: str = "None",
) -> "ReplicationGroup":
"""
Deletes an existing replication group. By default, this operation deletes the
entire replication group, including the primary/primaries and all of the read
replicas. If the replication group has only one primary, you can optionally
delete only the read replicas, while retaining the primary by setting
``RetainPrimaryCluster=true``.
Args:
ReplicationGroupId: The identifier for the cluster to be deleted. This parameter is not case sensitive.
Keyword Args:
RetainPrimaryCluster: If set to ``true``, all of the read replicas are deleted, but the primary node is retained.
FinalSnapshotIdentifier: The name of a final node group (shard) snapshot. ElastiCache creates the snapshot from the
primary node in the cluster, rather than one of the replicas; this is to ensure that it captures the freshest data.
After the final snapshot is taken, the replication group is immediately deleted.
"""
args: dict[str, Any] = dict(
ReplicationGroupId=self.serialize(ReplicationGroupId),
RetainPrimaryCluster=self.serialize(RetainPrimaryCluster),
FinalSnapshotIdentifier=self.serialize(FinalSnapshotIdentifier),
)
_response = self.client.delete_replication_group(
**{k: v for k, v in args.items() if v is not None}
)
response = DeleteReplicationGroupResult(**_response)
return cast("ReplicationGroup", response.ElasticacheReplicationGroup)
[docs] def get(self, ReplicationGroupId: str) -> "ReplicationGroup | None":
"""
Returns information about a particular replication group. If no identifier is
specified, ``DescribeReplicationGroups`` returns information about all
replication groups.
Args:
ReplicationGroupId: The identifier for the replication group to be described. This parameter is not case sensitive.
"""
args: dict[str, Any] = dict(
ReplicationGroupId=self.serialize(ReplicationGroupId)
)
_response = self.client.describe_replication_groups(
**{k: v for k, v in args.items() if v is not None}
)
response = ReplicationGroupMessage(**_response)
if response and response.ReplicationGroups:
self.sessionize(response.ReplicationGroups[0])
return response.ReplicationGroups[0]
return None
[docs] def list(
self, *, ReplicationGroupId: "str | None" = None
) -> PrimaryBoto3ModelQuerySet:
"""
Returns information about a particular replication group. If no identifier is
specified, ``DescribeReplicationGroups`` returns information about all
replication groups.
Keyword Args:
ReplicationGroupId: The identifier for the replication group to be described. This parameter is not case sensitive.
"""
paginator = self.client.get_paginator("describe_replication_groups")
args: dict[str, Any] = dict(
ReplicationGroupId=self.serialize(ReplicationGroupId)
)
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 = ReplicationGroupMessage(**_response)
if response.ReplicationGroups:
results.extend(response.ReplicationGroups)
else:
if getattr(response, "NextToken", None):
continue
break
self.sessionize(results)
if results and isinstance(results[0], Boto3Model):
return PrimaryBoto3ModelQuerySet(results)
return results
[docs]class CacheParameterManager(ReadonlyBoto3ModelManager):
service_name: str = "elasticache"
[docs] def get(
self, CacheParameterGroupName: str, *, Source: "str | None" = None
) -> "CacheParameter | None":
"""
Returns the detailed parameter list for a particular cache parameter group.
Args:
CacheParameterGroupName: The name of a specific cache parameter group to return details for.
Keyword Args:
Source: The parameter types to return.
"""
args: dict[str, Any] = dict(
CacheParameterGroupName=self.serialize(CacheParameterGroupName),
Source=self.serialize(Source),
)
_response = self.client.describe_cache_parameters(
**{k: v for k, v in args.items() if v is not None}
)
response = CacheParameterGroupDetails(**_response)
if response and response.Parameters:
self.sessionize(response.Parameters[0])
return response.Parameters[0]
return None
[docs] def List(
self, CacheParameterGroupName: str, *, Source: "str | None" = None
) -> "builtins.list[CacheParameter]":
"""
Returns the detailed parameter list for a particular cache parameter group.
Args:
CacheParameterGroupName: The name of a specific cache parameter group to return details for.
Keyword Args:
Source: The parameter types to return.
"""
paginator = self.client.get_paginator("describe_cache_parameters")
args: dict[str, Any] = dict(
CacheParameterGroupName=self.serialize(CacheParameterGroupName),
Source=self.serialize(Source),
)
response_iterator = paginator.paginate(
**{k: v for k, v in args.items() if v is not None}
)
results: "builtins.list[CacheParameter]" = []
for _response in response_iterator:
response = CacheParameterGroupDetails(**_response)
if response.Parameters is not None:
results.extend(response.Parameters)
else:
break
self.sessionize(results)
return cast("builtins.list[CacheParameter]", results)
[docs]class ElastiCacheUserManager(Boto3ModelManager):
service_name: str = "elasticache"
[docs] def create(
self,
model: "ElastiCacheUser",
Passwords: "builtins.list[str] | None" = None,
NoPasswordRequired: "bool | None" = None,
AuthenticationMode: "AuthenticationMode | None" = None,
) -> "ElastiCacheUser":
"""
For Valkey engine version 7.2 onwards and Redis OSS 6.0 to 7.1: Creates a user. For more information, see `Using Role
Based Access Control (RBAC) <http://docs.aws.amazon.com/AmazonElastiCache/latest/dg/Clusters.RBAC.html>`_.
Args:
model: The :py:class:`User` to create.
Keyword Args:
Passwords: Passwords used for this user. You can create up to two passwords for each user.
NoPasswordRequired: Indicates a password is not required for this user.
AuthenticationMode: Specifies how to authenticate the user.
"""
data = model.model_dump(exclude_none=True, by_alias=True)
args = dict(
UserId=data.get("UserId"),
UserName=data.get("UserName"),
Engine=data.get("Engine"),
AccessString=data.get("AccessString"),
Passwords=self.serialize(Passwords),
NoPasswordRequired=self.serialize(NoPasswordRequired),
Tags=data.get("Tags"),
AuthenticationMode=self.serialize(AuthenticationMode),
)
_response = self.client.create_user(
**{k: v for k, v in args.items() if v is not None}
)
response = ElastiCacheUser(**_response)
self.sessionize(response)
return cast("ElastiCacheUser", response)
[docs] def update(
self,
model: "ElastiCacheUser",
AppendAccessString: "str | None" = None,
Passwords: "builtins.list[str] | None" = None,
NoPasswordRequired: "bool | None" = None,
AuthenticationMode: "AuthenticationMode | None" = None,
) -> "ElastiCacheUser":
"""
Changes user password(s) and/or access string.
Args:
model: The :py:class:`User` to update.
Keyword Args:
AppendAccessString: Adds additional user permissions to the access string.
Passwords: The passwords belonging to the user. You are allowed up to two.
NoPasswordRequired: Indicates no password is required for the user.
AuthenticationMode: Specifies how to authenticate the user.
"""
data = model.model_dump(exclude_none=True, by_alias=True)
args = dict(
UserId=data.get("UserId"),
AccessString=data.get("AccessString"),
AppendAccessString=self.serialize(AppendAccessString),
Passwords=self.serialize(Passwords),
NoPasswordRequired=self.serialize(NoPasswordRequired),
AuthenticationMode=self.serialize(AuthenticationMode),
Engine=data.get("Engine"),
)
_response = self.client.modify_user(
**{k: v for k, v in args.items() if v is not None}
)
response = ElastiCacheUser(**_response)
self.sessionize(response)
return cast("ElastiCacheUser", response)
[docs] def delete(self, UserId: str) -> "ElastiCacheUser":
"""
For Valkey engine version 7.2 onwards and Redis OSS 6.0 onwards: Deletes a user. The user will be removed from all user
groups and in turn removed from all replication groups. For more information, see `Using Role Based Access Control
(RBAC) <http://docs.aws.amazon.com/AmazonElastiCache/latest/dg/Clusters.RBAC.html>`_.
Args:
UserId: The ID of the user.
"""
args: dict[str, Any] = dict(UserId=self.serialize(UserId))
_response = self.client.delete_user(
**{k: v for k, v in args.items() if v is not None}
)
response = ElastiCacheUser(**_response)
return response
[docs] @elasticache_user_add_tags
def get(
self, UserId: str, *, Engine: "str | None" = None
) -> "ElastiCacheUser | None":
"""
Returns a list of users.
Args:
UserId: The ID of the user.
Keyword Args:
Engine: The engine.
"""
args: dict[str, Any] = dict(
Engine=self.serialize(Engine), UserId=self.serialize(UserId)
)
_response = self.client.describe_users(
**{k: v for k, v in args.items() if v is not None}
)
response = DescribeUsersResult(**_response)
if response and response.Users:
self.sessionize(response.Users[0])
return response.Users[0]
return None
[docs] @elasticache_users_get_add_tags
def list(
self,
*,
Engine: "str | None" = None,
UserId: "str | None" = None,
Filters: "builtins.list[Filter] | None" = None,
) -> PrimaryBoto3ModelQuerySet:
"""
Returns a list of users.
Keyword Args:
Engine: The engine.
UserId: The ID of the user.
Filters: Filter to determine the list of User IDs to return.
"""
paginator = self.client.get_paginator("describe_users")
args: dict[str, Any] = dict(
Engine=self.serialize(Engine),
UserId=self.serialize(UserId),
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 = DescribeUsersResult(**_response)
if response.Users:
results.extend(response.Users)
else:
if getattr(response, "NextToken", None):
continue
break
self.sessionize(results)
if results and isinstance(results[0], Boto3Model):
return PrimaryBoto3ModelQuerySet(results)
return results
[docs]class ElastiCacheUserGroupManager(Boto3ModelManager):
service_name: str = "elasticache"
[docs] def create(self, model: "ElastiCacheUserGroup") -> "ElastiCacheUserGroup":
"""
For Valkey engine version 7.2 onwards and Redis OSS 6.0 to 7.1: Creates a user group. For more information, see `Using
Role Based Access Control (RBAC) <http://docs.aws.amazon.com/AmazonElastiCache/latest/dg/Clusters.RBAC.html>`_
Args:
model: The :py:class:`UserGroup` to create.
"""
data = model.model_dump(exclude_none=True, by_alias=True)
args = dict(
UserGroupId=data.get("UserGroupId"),
Engine=data.get("Engine"),
UserIds=data.get("UserIds"),
Tags=data.get("Tags"),
)
_response = self.client.create_user_group(
**{k: v for k, v in args.items() if v is not None}
)
response = ElastiCacheUserGroup(**_response)
self.sessionize(response)
return cast("ElastiCacheUserGroup", response)
[docs] def update(
self,
model: "ElastiCacheUserGroup",
UserIdsToAdd: "builtins.list[str] | None" = None,
UserIdsToRemove: "builtins.list[str] | None" = None,
) -> "ElastiCacheUserGroup":
"""
Changes the list of users that belong to the user group.
Args:
model: The :py:class:`UserGroup` to update.
Keyword Args:
UserIdsToAdd: The list of user IDs to add to the user group.
UserIdsToRemove: The list of user IDs to remove from the user group.
"""
data = model.model_dump(exclude_none=True, by_alias=True)
args = dict(
UserGroupId=data.get("UserGroupId"),
UserIdsToAdd=self.serialize(UserIdsToAdd),
UserIdsToRemove=self.serialize(UserIdsToRemove),
Engine=data.get("Engine"),
)
_response = self.client.modify_user_group(
**{k: v for k, v in args.items() if v is not None}
)
response = ElastiCacheUserGroup(**_response)
self.sessionize(response)
return cast("ElastiCacheUserGroup", response)
[docs] def delete(self, UserGroupId: str) -> "ElastiCacheUserGroup":
"""
For Valkey engine version 7.2 onwards and Redis OSS 6.0 onwards: Deletes a user group. The user group must first be
disassociated from the replication group before it can be deleted. For more information, see `Using Role Based Access
Control (RBAC) <http://docs.aws.amazon.com/AmazonElastiCache/latest/dg/Clusters.RBAC.html>`_.
Args:
UserGroupId: The ID of the user group.
"""
args: dict[str, Any] = dict(UserGroupId=self.serialize(UserGroupId))
_response = self.client.delete_user_group(
**{k: v for k, v in args.items() if v is not None}
)
response = ElastiCacheUserGroup(**_response)
return response
[docs] @elasticache_user_group_add_tags
def get(self, UserGroupId: str) -> "ElastiCacheUserGroup | None":
"""
Returns a list of user groups.
Args:
UserGroupId: The ID of the user group.
"""
args: dict[str, Any] = dict(UserGroupId=self.serialize(UserGroupId))
_response = self.client.describe_user_groups(
**{k: v for k, v in args.items() if v is not None}
)
response = DescribeUserGroupsResult(**_response)
if response and response.UserGroups:
self.sessionize(response.UserGroups[0])
return response.UserGroups[0]
return None
[docs] @elasticache_user_groups_get_add_tags
def list(self, *, UserGroupId: "str | None" = None) -> PrimaryBoto3ModelQuerySet:
"""
Returns a list of user groups.
Keyword Args:
UserGroupId: The ID of the user group.
"""
paginator = self.client.get_paginator("describe_user_groups")
args: dict[str, Any] = dict(UserGroupId=self.serialize(UserGroupId))
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 = DescribeUserGroupsResult(**_response)
if response.UserGroups:
results.extend(response.UserGroups)
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 ElastiCacheEndpoint(Boto3Model):
"""
Represents the information required for client programs to connect to a cache node.
This value is read-only.
"""
Address: "str | None" = None
"""
The DNS hostname of the cache node.
"""
Port: "int | None" = None
"""
The port number that the cache engine is listening on.
"""
[docs]class CloudWatchLogsDestinationDetails(Boto3Model):
"""
The configuration details of the CloudWatch Logs destination.
"""
LogGroup: "str | None" = None
"""
The name of the CloudWatch Logs log group.
"""
[docs]class KinesisFirehoseDestinationDetails(Boto3Model):
"""
The configuration details of the Kinesis Data Firehose destination.
"""
DeliveryStream: "str | None" = None
"""
The name of the Kinesis Data Firehose delivery stream.
"""
[docs]class ElastiCacheDestinationDetails(Boto3Model):
"""
Configuration details of either a CloudWatch Logs destination or Kinesis Data
Firehose destination.
"""
CloudWatchLogsDetails: "CloudWatchLogsDestinationDetails | None" = None
"""
The configuration details of the CloudWatch Logs destination.
"""
KinesisFirehoseDetails: "KinesisFirehoseDestinationDetails | None" = None
"""
The configuration details of the Kinesis Data Firehose destination.
"""
[docs]class PendingLogDeliveryConfiguration(Boto3Model):
"""
The log delivery configurations being modified.
"""
LogType: "Literal['slow-log', 'engine-log'] | None" = None
"""
Refers to `slow-log <https://redis.io/commands/slowlog>`_ or engine-log..
"""
DestinationType: "Literal['cloudwatch-logs', 'kinesis-firehose'] | None" = None
"""
Returns the destination type, either CloudWatch Logs or Kinesis Data Firehose.
"""
DestinationDetails: "ElastiCacheDestinationDetails | None" = None
"""
Configuration details of either a CloudWatch Logs destination or Kinesis Data
Firehose destination.
"""
LogFormat: "Literal['text', 'json'] | None" = None
"""
Returns the log format, either JSON or TEXT.
"""
[docs]class ElastiCacheScaleConfig(Boto3Model):
"""
Configuration settings for horizontal or vertical scaling operations on Memcached
clusters.
"""
ScalePercentage: "int | None" = None
"""
The percentage by which to scale the Memcached cluster, either horizontally by
adding nodes or vertically by increasing resources.
"""
ScaleIntervalMinutes: "int | None" = None
"""
The time interval in seconds between scaling operations when performing gradual
scaling for a Memcached cluster.
"""
[docs]class ElastiCachePendingModifiedValues(Boto3Model):
"""
A group of settings that are applied to the cluster in the future, or that are
currently being applied.
"""
NumCacheNodes: "int | None" = None
"""
The new number of cache nodes for the cluster.
"""
CacheNodeIdsToRemove: "builtins.list[str] | None" = Field(default_factory=list)
"""
A list of cache node IDs that are being removed (or will be removed) from the
cluster.
A node ID is a 4-digit numeric identifier (0001, 0002, etc.).
"""
EngineVersion: "str | None" = None
"""
The new cache engine version that the cluster runs.
"""
CacheNodeType: "str | None" = None
"""
The cache node type that this cluster or replication group is scaled to.
"""
AuthTokenStatus: "Literal['SETTING', 'ROTATING'] | None" = None
"""
The auth token status.
"""
LogDeliveryConfigurations: "builtins.list[PendingLogDeliveryConfiguration] | None" = Field(
default_factory=list
)
"""
The log delivery configurations being modified.
"""
TransitEncryptionEnabled: "bool | None" = None
"""
A flag that enables in-transit encryption when set to true.
"""
TransitEncryptionMode: "Literal['preferred', 'required'] | None" = None
"""
A setting that allows you to migrate your clients to use in-transit encryption, with
no downtime.
"""
ScaleConfig: "ElastiCacheScaleConfig | None" = None
"""
The scaling configuration changes that are pending for the Memcached cluster.
"""
[docs]class ElastiCacheNotificationConfiguration(Boto3Model):
"""
Describes a notification topic and its status.
Notification topics are used for publishing ElastiCache events to subscribers using
Amazon Simple Notification Service (SNS).
"""
TopicArn: "str | None" = None
"""
The Amazon Resource Name (ARN) that identifies the topic.
"""
TopicStatus: "str | None" = None
"""
The current state of the topic.
"""
[docs]class CacheSecurityGroupMembership(Boto3Model):
"""
Represents a cluster's status within a particular cache security group.
"""
CacheSecurityGroupName: "str | None" = None
"""
The name of the cache security group.
"""
Status: "str | None" = None
"""
The membership status in the cache security group.
The status changes when a cache security group is modified, or when the cache
security groups assigned to a cluster are modified.
"""
[docs]class CacheParameterGroupStatus(Boto3Model):
"""
Status of the cache parameter group.
"""
CacheParameterGroupName: "str | None" = None
"""
The name of the cache parameter group.
"""
ParameterApplyStatus: "str | None" = None
"""
The status of parameter updates.
"""
CacheNodeIdsToReboot: "builtins.list[str] | None" = Field(default_factory=list)
"""
A list of the cache node IDs which need to be rebooted for parameter changes to be
applied.
A node ID is a numeric identifier (0001, 0002, etc.).
"""
[docs]class CacheNode(Boto3Model):
"""Represents an individual cache node within a cluster. Each cache node runs its own instance of the cluster's protocol-
compliant caching software - either Memcached, Valkey or Redis OSS.
The following node types are supported by ElastiCache. Generally speaking, the current generation types provide more
memory and computational power at lower cost when compared to their equivalent previous generation counterparts.
* General purpose:
+ Current generation:
**M7g node types**: ``cache.m7g.large``, ``cache.m7g.xlarge``, ``cache.m7g.2xlarge``, ``cache.m7g.4xlarge``,
``cache.m7g.8xlarge``, ``cache.m7g.12xlarge``, ``cache.m7g.16xlarge``
For region availability, see `Supported Node Types <https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/CacheNodes.S
upportedTypes.html#CacheNodes.SupportedTypesByRegion>`_
**M6g node types** (available only for Redis OSS engine version 5.0.6 onward and for Memcached engine version 1.5.16
onward): ``cache.m6g.large``, ``cache.m6g.xlarge``, ``cache.m6g.2xlarge``, ``cache.m6g.4xlarge``, ``cache.m6g.8xlarge``,
``cache.m6g.12xlarge``, ``cache.m6g.16xlarge``
**M5 node types:** ``cache.m5.large``, ``cache.m5.xlarge``, ``cache.m5.2xlarge``, ``cache.m5.4xlarge``,
``cache.m5.12xlarge``, ``cache.m5.24xlarge``
**M4 node types:** ``cache.m4.large``, ``cache.m4.xlarge``, ``cache.m4.2xlarge``, ``cache.m4.4xlarge``,
``cache.m4.10xlarge``
**T4g node types** (available only for Redis OSS engine version 5.0.6 onward and Memcached engine version 1.5.16
onward): ``cache.t4g.micro``, ``cache.t4g.small``, ``cache.t4g.medium``
**T3 node types:** ``cache.t3.micro``, ``cache.t3.small``, ``cache.t3.medium``
**T2 node types:** ``cache.t2.micro``, ``cache.t2.small``, ``cache.t2.medium``
+ Previous generation: (not recommended. Existing clusters are still supported but creation of new clusters is not
supported for these types.)
**T1 node types:** ``cache.t1.micro``
**M1 node types:** ``cache.m1.small``, ``cache.m1.medium``, ``cache.m1.large``, ``cache.m1.xlarge``
**M3 node types:** ``cache.m3.medium``, ``cache.m3.large``, ``cache.m3.xlarge``, ``cache.m3.2xlarge``
* Compute optimized:
+ Previous generation: (not recommended. Existing clusters are still supported but creation of new clusters is not
supported for these types.)
**C1 node types:** ``cache.c1.xlarge``
* Memory optimized:
+ Current generation:
**R7g node types**: ``cache.r7g.large``, ``cache.r7g.xlarge``, ``cache.r7g.2xlarge``, ``cache.r7g.4xlarge``,
``cache.r7g.8xlarge``, ``cache.r7g.12xlarge``, ``cache.r7g.16xlarge``
For region availability, see `Supported Node Types <https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/CacheNodes.S
upportedTypes.html#CacheNodes.SupportedTypesByRegion>`_
**R6g node types** (available only for Redis OSS engine version 5.0.6 onward and for Memcached engine version 1.5.16
onward): ``cache.r6g.large``, ``cache.r6g.xlarge``, ``cache.r6g.2xlarge``, ``cache.r6g.4xlarge``, ``cache.r6g.8xlarge``,
``cache.r6g.12xlarge``, ``cache.r6g.16xlarge``
**R5 node types:** ``cache.r5.large``, ``cache.r5.xlarge``, ``cache.r5.2xlarge``, ``cache.r5.4xlarge``,
``cache.r5.12xlarge``, ``cache.r5.24xlarge``
**R4 node types:** ``cache.r4.large``, ``cache.r4.xlarge``, ``cache.r4.2xlarge``, ``cache.r4.4xlarge``,
``cache.r4.8xlarge``, ``cache.r4.16xlarge``
+ Previous generation: (not recommended. Existing clusters are still supported but creation of new clusters is not
supported for these types.)
**M2 node types:** ``cache.m2.xlarge``, ``cache.m2.2xlarge``, ``cache.m2.4xlarge``
**R3 node types:** ``cache.r3.large``, ``cache.r3.xlarge``, ``cache.r3.2xlarge``, ``cache.r3.4xlarge``,
``cache.r3.8xlarge``
**Additional node type info**
* All current generation instance types are created in Amazon VPC by default.
* Valkey or Redis OSS append-only files (AOF) are not supported for T1 or T2 instances.
* Valkey or Redis OSS Multi-AZ with automatic failover is not supported on T1 instances.
* The configuration variables ``appendonly`` and ``appendfsync`` are not supported on Valkey, or on Redis OSS version
2.8.22 and later.
"""
CacheNodeId: "str | None" = None
"""
The cache node identifier.
A node ID is a numeric identifier (0001, 0002, etc.). The combination of cluster ID
and node ID uniquely identifies every cache node used in a customer's Amazon
account.
"""
CacheNodeStatus: "str | None" = None
"""
The current state of this cache node, one of the following values: ``available``,
``creating``, ``rebooting``, or ``deleting``.
"""
CacheNodeCreateTime: "datetime | None" = None
"""
The date and time when the cache node was created.
"""
Endpoint: "ElastiCacheEndpoint | None" = None
"""
The hostname for connecting to this cache node.
"""
ParameterGroupStatus: "str | None" = None
"""
The status of the parameter group applied to this cache node.
"""
SourceCacheNodeId: "str | None" = None
"""
The ID of the primary node to which this read replica node is synchronized.
If this field is empty, this node is not associated with a primary cluster.
"""
CustomerAvailabilityZone: "str | None" = None
"""
The Availability Zone where this node was created and now resides.
"""
CustomerOutpostArn: "str | None" = None
"""
The customer outpost ARN of the cache node.
"""
[docs]class SecurityGroupMembership(Boto3Model):
"""
Represents a single cache security group and its status.
"""
SecurityGroupId: "str | None" = None
"""
The identifier of the cache security group.
"""
Status: "str | None" = None
"""
The status of the cache security group membership.
The status changes whenever a cache security group is modified, or when the cache
security groups assigned to a cluster are modified.
"""
[docs]class LogDeliveryConfiguration(Boto3Model):
"""
Returns the destination, format and type of the logs.
"""
LogType: "Literal['slow-log', 'engine-log'] | None" = None
"""
Refers to `slow-log <https://redis.io/commands/slowlog>`_ or engine-log.
"""
DestinationType: "Literal['cloudwatch-logs', 'kinesis-firehose'] | None" = None
"""
Returns the destination type, either ``cloudwatch-logs`` or ``kinesis-firehose``.
"""
DestinationDetails: "ElastiCacheDestinationDetails | None" = None
"""
Configuration details of either a CloudWatch Logs destination or Kinesis Data
Firehose destination.
"""
LogFormat: "Literal['text', 'json'] | None" = None
"""
Returns the log format, either JSON or TEXT.
"""
Status: "Literal['active', 'enabling', 'modifying', 'disabling', 'error'] | None" = None
"""
Returns the log delivery configuration status.
Values are one of ``enabling`` | ``disabling`` | ``modifying`` |
``active`` | ``error``
"""
Message: "str | None" = None
"""
Returns an error message for the log delivery configuration.
"""
[docs]class CacheCluster(CacheClusterModelMixin, PrimaryBoto3Model):
"""
Contains all of the attributes of a specific cluster.
"""
manager_class: ClassVar[type[Boto3ModelManager]] = CacheClusterManager
NumCacheNodes: "int | None" = 1
"""
The number of cache nodes in the cluster.
"""
CacheClusterId: "str | None" = None
"""
The user-supplied identifier of the cluster.
This identifier is a unique key that identifies a cluster.
"""
ConfigurationEndpoint: ElastiCacheEndpoint = Field(default=None, frozen=True)
"""
Represents a Memcached cluster endpoint which can be used by an application to
connect to any node in the cluster.
The
configuration endpoint will always have ``.cfg`` in it.
"""
ClientDownloadLandingPage: str = Field(default=None, frozen=True)
"""
The URL of the web page where you can download the latest ElastiCache client
library.
"""
CacheNodeType: "str | None" = None
"""
The name of the compute and memory capacity node type for the cluster.
"""
Engine: "str | None" = None
"""
The name of the cache engine (``memcached`` or ``redis``) to be used for this
cluster.
"""
EngineVersion: "str | None" = None
"""
The version of the cache engine that is used in this cluster.
"""
CacheClusterStatus: str = Field(default=None, frozen=True)
"""
The current state of this cluster, one of the following values: ``available``,
``creating``, ``deleted``, ``deleting``, ``incompatible-network``, ``modifying``,
``rebooting cluster nodes``, ``restore-failed``, or ``snapshotting``.
"""
PreferredAvailabilityZone: "str | None" = None
"""
The name of the Availability Zone in which the cluster is located or "Multiple" if
the cache nodes are located in different Availability Zones.
"""
PreferredOutpostArn: "str | None" = None
"""
The outpost ARN in which the cache cluster is created.
"""
CacheClusterCreateTime: datetime = Field(default=None, frozen=True)
"""
The date and time when the cluster was created.
"""
PreferredMaintenanceWindow: "str | None" = None
"""
Specifies the weekly time range during which maintenance on the cluster is
performed.
It is specified as a range in the format ddd:hh24:mi-ddd:hh24:mi (24H Clock UTC).
The minimum maintenance window is a 60 minute period.
"""
PendingModifiedValues: ElastiCachePendingModifiedValues = Field(
default=None, frozen=True
)
"""
A group of settings that are applied to the cluster in the future, or that are
currently being applied.
"""
NotificationConfiguration: ElastiCacheNotificationConfiguration = Field(
default=None, frozen=True
)
"""
Describes a notification topic and its status.
Notification topics are used for publishing ElastiCache events to subscribers using
Amazon Simple Notification Service (SNS).
"""
CacheSecurityGroups: "builtins.list[CacheSecurityGroupMembership]" = Field(
default_factory=list, frozen=True
)
"""
A list of cache security group elements, composed of name and status sub-elements.
"""
CacheParameterGroup: CacheParameterGroupStatus = Field(default=None, frozen=True)
"""
Status of the cache parameter group.
"""
CacheSubnetGroupName: "str | None" = None
"""
The name of the cache subnet group associated with the cluster.
"""
CacheNodes: "builtins.list[CacheNode]" = Field(default_factory=list, frozen=True)
"""
A list of cache nodes that are members of the cluster.
"""
AutoMinorVersionUpgrade: "bool | None" = None
"""
If you are running Valkey or Redis OSS engine version 6.0 or later, set this
parameter to yes if you want to opt-in to the next auto minor version upgrade
campaign.
This parameter is disabled for previous versions.
"""
SecurityGroups: "builtins.list[SecurityGroupMembership]" = Field(
default_factory=list, frozen=True
)
"""
A list of VPC Security Groups associated with the cluster.
"""
ReplicationGroupId: "str | None" = None
"""
The replication group to which this cluster belongs.
If this field is empty, the cluster is not associated with any replication group.
"""
SnapshotRetentionLimit: "int | None" = None
"""
The number of days for which ElastiCache retains automatic cluster snapshots before
deleting them.
For example, if you
set ``SnapshotRetentionLimit`` to 5, a snapshot that was taken today is retained for 5 days before being deleted.
"""
SnapshotWindow: "str | None" = None
"""
The daily time range (in UTC) during which ElastiCache begins taking a daily
snapshot of your cluster.
"""
AuthTokenEnabled: bool = Field(default=None, frozen=True)
"""
A flag that enables using an ``AuthToken`` (password) when issuing Valkey or Redis
OSS commands.
"""
AuthTokenLastModifiedDate: datetime = Field(default=None, frozen=True)
"""
The date the auth token was last modified.
"""
TransitEncryptionEnabled: "bool | None" = None
"""
A flag that enables in-transit encryption when set to ``true``.
"""
AtRestEncryptionEnabled: bool = Field(default=None, frozen=True)
"""
A flag that enables encryption at-rest when set to ``true``.
"""
ARN: str = Field(default=None, frozen=True)
"""
The ARN (Amazon Resource Name) of the cache cluster.
"""
ReplicationGroupLogDeliveryEnabled: bool = Field(default=None, frozen=True)
"""
A boolean value indicating whether log delivery is enabled for the replication
group.
"""
LogDeliveryConfigurations: "builtins.list[LogDeliveryConfiguration] | None" = Field(
default_factory=list
)
"""
Returns the destination, format and type of the logs.
"""
NetworkType: "Literal['ipv4', 'ipv6', 'dual_stack'] | None" = None
"""
Must be either ``ipv4`` | ``ipv6`` | ``dual_stack``.
IPv6 is supported for workloads using Valkey 7.2 and above, Redis OSS engine version
6.2 7.1 or Memcached engine version 1.6.6 and above on all instances built on the
`Nitro system <http://aws.amazon.com/ec2/nitro/>`_.
"""
IpDiscovery: "Literal['ipv4', 'ipv6'] | None" = None
"""
The network type associated with the cluster, either ``ipv4`` | ``ipv6``.
IPv6 is supported for workloads using Valkey
7.2 and above, Redis OSS engine version 6.2 to 7.1 or Memcached engine version 1.6.6 and above on all instances built on
the `Nitro system <http://aws.amazon.com/ec2/nitro/>`_.
"""
TransitEncryptionMode: Literal["preferred", "required"] = Field(
default=None, frozen=True
)
"""
A setting that allows you to migrate your clients to use in-transit encryption, with
no downtime.
"""
@property
def pk(self) -> str | None:
"""
Return the primary key of the model. This is the value of the :py:attr:`ARN`
attribute.
Returns:
The primary key of the model instance.
"""
return self.ARN
@property
def arn(self) -> str | None:
"""
Return the ARN of the model. This is the value of the :py:attr:`ARN`
attribute.
Returns:
The ARN of the model instance.
"""
return self.ARN
@property
def name(self) -> str | None:
"""
Return the name of the model. This is the value of the
:py:attr:`CacheClusterId` attribute.
Returns:
The name of the model instance.
"""
return self.CacheClusterId
def __hash__(self) -> int:
"""
Return the hash of the model.
This is the value of the
:py:attr:`ARN` attribute.
"""
return hash(self.ARN)
@cached_property
def parameter_group(self) -> "CacheParameterGroup | None":
"""
Return the cache parameter group associated with this 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(
{
"CacheParameterGroupName": self.CacheParameterGroup.CacheParameterGroupName,
}
)
except AttributeError:
return None
return CacheParameterGroup.objects.using(self.session).get(**pk) # type: ignore[arg-type]
@cached_property
def subnet_group(self) -> "CacheSubnetGroup | None":
"""
Return the cache subnet group associated with this 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(
{
"CacheSubnetGroupName": self.CacheSubnetGroupName,
}
)
except AttributeError:
return None
return CacheSubnetGroup.objects.using(self.session).get(**pk) # type: ignore[arg-type]
[docs]class ParameterNameValue(Boto3Model):
"""
Describes a name-value pair that is used to update the value of a parameter.
"""
ParameterName: "str | None" = None
"""
The name of the parameter.
"""
ParameterValue: "str | None" = None
"""
The value of the parameter.
"""
[docs]class CacheParameterGroup(PrimaryBoto3Model):
"""
Represents the output of a ``CreateCacheParameterGroup`` operation.
"""
manager_class: ClassVar[type[Boto3ModelManager]] = CacheParameterGroupManager
IsGlobal: "bool | None"
"""
Indicates whether the parameter group is associated with a Global datastore.
"""
CacheParameterGroupName: "str | None" = None
"""
The name of the cache parameter group.
"""
CacheParameterGroupFamily: "str | None" = None
"""
The name of the cache parameter group family that this cache parameter group is
compatible with.
"""
Description: "str | None" = None
"""
The description for this cache parameter group.
"""
ARN: str = Field(default=None, frozen=True)
"""
The ARN (Amazon Resource Name) of the cache parameter group.
"""
@property
def pk(self) -> str | None:
"""
Return the primary key of the model. This is the value of the :py:attr:`ARN`
attribute.
Returns:
The primary key of the model instance.
"""
return self.ARN
@property
def arn(self) -> str | None:
"""
Return the ARN of the model. This is the value of the :py:attr:`ARN`
attribute.
Returns:
The ARN of the model instance.
"""
return self.ARN
@property
def name(self) -> str | None:
"""
Return the name of the model. This is the value of the
:py:attr:`CacheParameterGroupName` attribute.
Returns:
The name of the model instance.
"""
return self.CacheParameterGroupName
def __hash__(self) -> int:
"""
Return the hash of the model.
This is the value of the
:py:attr:`ARN` attribute.
"""
return hash(self.ARN)
[docs] def parameters(
self, Source: "str | None" = "user"
) -> "builtins.list[CacheParameter]":
"""
Return the parameters for the cache parameter group associated with this
cluster.
Keyword Args:
Source: The parameter types to return.
"""
return (
cast("CacheParameterGroupManager", self.objects) # type: ignore[attr-defined]
.using(self.session)
.parameters(cast("str", self.CacheParameterGroupName), Source=Source)
)
[docs] def reset(
self,
ResetAllParameters: "bool | None" = None,
ParameterNameValues: "builtins.list[ParameterNameValue] | None" = {},
) -> "str":
"""
Reset the cache parameter group associated with this cluster to the default
values.
Keyword Args:
ResetAllParameters: If ``true``, all parameters in the cache parameter group are reset to their default values. If
``false``, only the parameters listed by ``ParameterNameValues`` are reset to their default values.
ParameterNameValues: An array of parameter names to reset to their default values. If ``ResetAllParameters`` is
``true``, do not use ``ParameterNameValues``. If ``ResetAllParameters`` is ``false``, you must specify the name of
at least one parameter to reset.
"""
return (
cast("CacheParameterGroupManager", self.objects) # type: ignore[attr-defined]
.using(self.session)
.reset(
cast("str", self.CacheParameterGroupName),
ResetAllParameters=ResetAllParameters,
ParameterNameValues=ParameterNameValues,
)
)
[docs]class CacheSubnetGroup(PrimaryBoto3Model):
"""Represents the output of one of the following operations:
* ``CreateCacheSubnetGroup``
* ``ModifyCacheSubnetGroup``
"""
manager_class: ClassVar[type[Boto3ModelManager]] = CacheSubnetGroupManager
CacheSubnetGroupName: "str | None" = None
"""
The name of the cache subnet group.
"""
CacheSubnetGroupDescription: "str | None" = None
"""
The description of the cache subnet group.
"""
VpcId: str = Field(default=None, frozen=True)
"""
The Amazon Virtual Private Cloud identifier (VPC ID) of the cache subnet group.
"""
Subnets: "builtins.list[Subnet]" = Field(default_factory=list, frozen=True)
"""
A list of subnets associated with the cache subnet group.
"""
ARN: str = Field(default=None, frozen=True)
"""
The ARN (Amazon Resource Name) of the cache subnet group.
"""
SupportedNetworkTypes: "builtins.list[Literal['ipv4', 'ipv6', 'dual_stack']]" = (
Field(default_factory=list, frozen=True)
)
"""
Either ``ipv4`` | ``ipv6`` | ``dual_stack``.
IPv6 is supported for workloads using Valkey 7.2 and above, Redis OSS engine version
6.2 to 7.1 or Memcached engine version 1.6.6 and above on all instances built on the
`Nitro system <http://aws.amazon.com/ec2/nitro/>`_.
"""
@property
def pk(self) -> str | None:
"""
Return the primary key of the model. This is the value of the :py:attr:`ARN`
attribute.
Returns:
The primary key of the model instance.
"""
return self.ARN
@property
def arn(self) -> str | None:
"""
Return the ARN of the model. This is the value of the :py:attr:`ARN`
attribute.
Returns:
The ARN of the model instance.
"""
return self.ARN
@property
def name(self) -> str | None:
"""
Return the name of the model. This is the value of the
:py:attr:`CacheSubnetGroupName` attribute.
Returns:
The name of the model instance.
"""
return self.CacheSubnetGroupName
def __hash__(self) -> int:
"""
Return the hash of the model.
This is the value of the
:py:attr:`ARN` attribute.
"""
return hash(self.ARN)
@cached_property
def ec2_subnets(self) -> "list[Subnet] | None":
"""
Return the EC2 subnets associated with this cache subnet group.
.. 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(
{
"SubnetIds": self.SubnetIds,
}
)
except AttributeError:
return []
return Subnet.objects.using(self.session).list(**pk) # type: ignore[arg-type]
[docs]class EC2SecurityGroup(Boto3Model):
"""
Provides ownership and status information for an Amazon EC2 security group.
"""
Status: "str | None" = None
"""
The status of the Amazon EC2 security group.
"""
EC2SecurityGroupName: "str | None" = None
"""
The name of the Amazon EC2 security group.
"""
EC2SecurityGroupOwnerId: "str | None" = None
"""
The Amazon account ID of the Amazon EC2 security group owner.
"""
[docs]class CacheSecurityGroup(PrimaryBoto3Model):
"""Represents the output of one of the following operations:
* ``AuthorizeCacheSecurityGroupIngress``
* ``CreateCacheSecurityGroup``
* ``RevokeCacheSecurityGroupIngress``
"""
manager_class: ClassVar[type[Boto3ModelManager]] = CacheSecurityGroupManager
OwnerId: str = Field(default=None, frozen=True)
"""
The Amazon account ID of the cache security group owner.
"""
CacheSecurityGroupName: "str | None" = None
"""
The name of the cache security group.
"""
Description: "str | None" = None
"""
The description of the cache security group.
"""
EC2SecurityGroups: "builtins.list[EC2SecurityGroup]" = Field(
default_factory=list, frozen=True
)
"""
A list of Amazon EC2 security groups that are associated with this cache security
group.
"""
ARN: str = Field(default=None, frozen=True)
"""
The ARN of the cache security group,
"""
@property
def pk(self) -> str | None:
"""
Return the primary key of the model. This is the value of the :py:attr:`ARN`
attribute.
Returns:
The primary key of the model instance.
"""
return self.ARN
@property
def arn(self) -> str | None:
"""
Return the ARN of the model. This is the value of the :py:attr:`ARN`
attribute.
Returns:
The ARN of the model instance.
"""
return self.ARN
@property
def name(self) -> str | None:
"""
Return the name of the model. This is the value of the
:py:attr:`CacheSecurityGroupName` attribute.
Returns:
The name of the model instance.
"""
return self.CacheSecurityGroupName
def __hash__(self) -> int:
"""
Return the hash of the model.
This is the value of the
:py:attr:`ARN` attribute.
"""
return hash(self.ARN)
@cached_property
def ec2_security_groups(self) -> "list[SecurityGroup] | None":
"""
Return the EC2 security groups associated with this cache security group.
.. 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(
{
"EC2SecurityGroups": self.EC2SecurityGroups,
}
)
except AttributeError:
return []
return SecurityGroup.objects.using(self.session).list(**pk) # type: ignore[arg-type]
[docs] def authorize_ingress(
self, EC2SecurityGroupName: str, EC2SecurityGroupOwnerId: str
) -> "CacheSecurityGroup":
"""
Authorize ingress to the cache security group associated with this cluster.
Args:
EC2SecurityGroupName: The Amazon EC2 security group to be authorized for ingress to the cache security group.
EC2SecurityGroupOwnerId: The Amazon account number of the Amazon EC2 security group owner. Note that this is not the
same thing as an Amazon access key ID - you must provide a valid Amazon account number for this parameter.
"""
return (
cast("CacheSecurityGroupManager", self.objects) # type: ignore[attr-defined]
.using(self.session)
.authorize_ingress(
cast("str", self.CacheSecurityGroupName),
EC2SecurityGroupName,
EC2SecurityGroupOwnerId,
)
)
[docs] def revoke_ingress(
self, EC2SecurityGroupName: str, EC2SecurityGroupOwnerId: str
) -> "CacheSecurityGroup":
"""
Authorize ingress to the cache security group associated with this cluster.
Args:
EC2SecurityGroupName: The Amazon EC2 security group to be authorized for ingress to the cache security group.
EC2SecurityGroupOwnerId: The Amazon account number of the Amazon EC2 security group owner. Note that this is not the
same thing as an Amazon access key ID - you must provide a valid Amazon account number for this parameter.
"""
return (
cast("CacheSecurityGroupManager", self.objects) # type: ignore[attr-defined]
.using(self.session)
.authorize_ingress(
cast("str", self.CacheSecurityGroupName),
EC2SecurityGroupName,
EC2SecurityGroupOwnerId,
)
)
[docs]class ElastiCacheGlobalReplicationGroupInfo(Boto3Model):
"""
The name of the Global datastore and role of this replication group in the Global
datastore.
"""
GlobalReplicationGroupId: "str | None" = None
"""
The name of the Global datastore.
"""
GlobalReplicationGroupMemberRole: "str | None" = None
"""
The role of the replication group in a Global datastore.
Can be primary or secondary.
"""
[docs]class ElastiCacheSlotMigration(Boto3Model):
"""
Represents the progress of an online resharding operation.
"""
ProgressPercentage: "float | None" = None
"""
The percentage of the slot migration that is complete.
"""
[docs]class ReshardingStatus(Boto3Model):
"""
The status of an online resharding operation.
"""
SlotMigration: "ElastiCacheSlotMigration | None" = None
"""
Represents the progress of an online resharding operation.
"""
[docs]class UserGroupsUpdateStatus(Boto3Model):
"""
The status of the user group update.
"""
UserGroupIdsToAdd: "builtins.list[str] | None" = Field(default_factory=list)
"""
The ID of the user group to add.
"""
UserGroupIdsToRemove: "builtins.list[str] | None" = Field(default_factory=list)
"""
The ID of the user group to remove.
"""
[docs]class ReplicationGroupPendingModifiedValues(Boto3Model):
"""
The settings to be applied to the Valkey or Redis OSS replication group, either
immediately or during the next maintenance window.
"""
PrimaryClusterId: "str | None" = None
"""
The primary cluster ID that is applied immediately (if ``--apply-immediately`` was
specified), or during the next maintenance window.
"""
AutomaticFailoverStatus: "Literal['enabled', 'disabled'] | None" = None
"""
Indicates the status of automatic failover for this Valkey or Redis OSS replication
group.
"""
Resharding: "ReshardingStatus | None" = None
"""
The status of an online resharding operation.
"""
AuthTokenStatus: "Literal['SETTING', 'ROTATING'] | None" = None
"""
The auth token status.
"""
UserGroups: "UserGroupsUpdateStatus | None" = None
"""
The user group being modified.
"""
LogDeliveryConfigurations: "builtins.list[PendingLogDeliveryConfiguration] | None" = Field(
default_factory=list
)
"""
The log delivery configurations being modified.
"""
TransitEncryptionEnabled: "bool | None" = None
"""
A flag that enables in-transit encryption when set to true.
"""
TransitEncryptionMode: "Literal['preferred', 'required'] | None" = None
"""
A setting that allows you to migrate your clients to use in-transit encryption, with
no downtime.
"""
ClusterMode: "Literal['enabled', 'disabled', 'compatible'] | None" = None
"""
Enabled or Disabled.
To modify cluster mode from Disabled to Enabled, you must first set the cluster mode
to Compatible. Compatible mode allows your Valkey or Redis OSS clients to connect
using both cluster mode enabled and cluster mode disabled. After you migrate all
Valkey or Redis OSS clients to use cluster mode enabled, you can then complete
cluster mode configuration and set the cluster mode to Enabled.
"""
[docs]class NodeGroupMember(Boto3Model):
"""
Represents a single node within a node group (shard).
"""
CacheClusterId: "str | None" = None
"""
The ID of the cluster to which the node belongs.
"""
CacheNodeId: "str | None" = None
"""
The ID of the node within its cluster.
A node ID is a numeric identifier (0001, 0002, etc.).
"""
ReadEndpoint: "ElastiCacheEndpoint | None" = None
"""
The information required for client programs to connect to a node for read
operations.
The read endpoint is only applicable on Valkey or Redis OSS (cluster mode disabled)
clusters.
"""
PreferredAvailabilityZone: "str | None" = None
"""
The name of the Availability Zone in which the node is located.
"""
PreferredOutpostArn: "str | None" = None
"""
The outpost ARN of the node group member.
"""
CurrentRole: "str | None" = None
"""
The role that is currently assigned to the node - ``primary`` or ``replica``. This member is only applicable for Valkey
or Redis OSS (cluster mode disabled) replication groups.
"""
[docs]class NodeGroup(Boto3Model):
"""
Represents a collection of cache nodes in a replication group.
One node in the node group is the read/write primary node. All the other nodes are
read-only Replica nodes.
"""
NodeGroupId: "str | None" = None
"""
The identifier for the node group (shard).
A Valkey or Redis OSS (cluster mode disabled) replication group contains only 1 node
group; therefore, the node group ID is 0001. A Valkey or Redis OSS (cluster mode
enabled) replication group contains 1 to 90 node groups numbered 0001 to 0090.
Optionally, the user can provide the id for a node group.
"""
Status: "str | None" = None
"""
The current state of this replication group - ``creating``, ``available``, ``modifying``, ``deleting``.
"""
PrimaryEndpoint: "ElastiCacheEndpoint | None" = None
"""
The endpoint of the primary node in this node group (shard).
"""
ReaderEndpoint: "ElastiCacheEndpoint | None" = None
"""
The endpoint of the replica nodes in this node group (shard).
This value is read-only.
"""
Slots: "str | None" = None
"""
The keyspace for this node group (shard).
"""
NodeGroupMembers: "builtins.list[NodeGroupMember] | None" = Field(
default_factory=list
)
"""
A list containing information about individual nodes within the node group (shard).
"""
[docs]class ReplicationGroup(ReplicationGroupModelMixin, PrimaryBoto3Model):
"""
Contains all of the attributes of a specific Valkey or Redis OSS replication group.
"""
manager_class: ClassVar[type[Boto3ModelManager]] = ReplicationGroupManager
MultiAZ: "Literal['enabled', 'disabled'] | None"
"""
A flag indicating if you have Multi-AZ enabled to enhance fault tolerance.
For more information, see `Minimizing
Downtime: Multi-AZ <http://docs.aws.amazon.com/AmazonElastiCache/latest/dg/AutoFailover.html>`_
"""
ReplicationGroupId: "str | None" = None
"""
The identifier for the replication group.
"""
Description: str = Field(default=None, frozen=True)
"""
The user supplied description of the replication group.
"""
GlobalReplicationGroupInfo: ElastiCacheGlobalReplicationGroupInfo = Field(
default=None, frozen=True
)
"""
The name of the Global datastore and role of this replication group in the Global
datastore.
"""
Status: str = Field(default=None, frozen=True)
"""
The current state of this replication group - ``creating``, ``available``, ``modifying``, ``deleting``, ``create-
failed``, ``snapshotting``.
"""
PendingModifiedValues: ReplicationGroupPendingModifiedValues = Field(
default=None, frozen=True
)
"""
A group of settings to be applied to the replication group, either immediately or
during the next maintenance window.
"""
MemberClusters: "builtins.list[str]" = Field(default_factory=list, frozen=True)
"""
The names of all the cache clusters that are part of this replication group.
"""
NodeGroups: "builtins.list[NodeGroup]" = Field(default_factory=list, frozen=True)
"""
A list of node groups in this replication group.
For Valkey or Redis OSS (cluster mode disabled) replication groups, this is a
single-element list. For Valkey or Redis OSS (cluster mode enabled) replication
groups, the list contains an entry for each node group (shard).
"""
SnapshottingClusterId: "str | None" = None
"""
The cluster ID that is used as the daily snapshot source for the replication group.
"""
AutomaticFailover: Literal["enabled", "disabled", "enabling", "disabling"] = Field(
default=None, frozen=True
)
"""
Indicates the status of automatic failover for this Valkey or Redis OSS replication
group.
"""
ConfigurationEndpoint: ElastiCacheEndpoint = Field(default=None, frozen=True)
"""
The configuration endpoint for this replication group.
Use the configuration endpoint to connect to this replication group.
"""
SnapshotRetentionLimit: "int | None" = None
"""
The number of days for which ElastiCache retains automatic cluster snapshots before
deleting them.
For example, if you
set ``SnapshotRetentionLimit`` to 5, a snapshot that was taken today is retained for 5 days before being deleted.
"""
SnapshotWindow: "str | None" = None
"""
The daily time range (in UTC) during which ElastiCache begins taking a daily
snapshot of your node group (shard).
"""
ClusterEnabled: bool = Field(default=None, frozen=True)
"""
A flag indicating whether or not this replication group is cluster enabled; i.e.,
whether its data can be partitioned across multiple shards (API/CLI: node groups).
"""
CacheNodeType: "str | None" = None
"""
The name of the compute and memory capacity node type for each node in the
replication group.
"""
AuthTokenEnabled: bool = Field(default=None, frozen=True)
"""
A flag that enables using an ``AuthToken`` (password) when issuing Valkey or Redis
OSS commands.
"""
AuthTokenLastModifiedDate: datetime = Field(default=None, frozen=True)
"""
The date the auth token was last modified.
"""
TransitEncryptionEnabled: "bool | None" = None
"""
A flag that enables in-transit encryption when set to ``true``.
"""
AtRestEncryptionEnabled: "bool | None" = None
"""
A flag that enables encryption at-rest when set to ``true``.
"""
MemberClustersOutpostArns: "builtins.list[str]" = Field(
default_factory=list, frozen=True
)
"""
The outpost ARNs of the replication group's member clusters.
"""
KmsKeyId: "str | None" = None
"""
The ID of the KMS key used to encrypt the disk in the cluster.
"""
ARN: str = Field(default=None, frozen=True)
"""
The ARN (Amazon Resource Name) of the replication group.
"""
UserGroupIds: "builtins.list[str] | None" = Field(default_factory=list)
"""
The ID of the user group associated to the replication group.
"""
LogDeliveryConfigurations: "builtins.list[LogDeliveryConfiguration] | None" = Field(
default_factory=list
)
"""
Returns the destination, format and type of the logs.
"""
ReplicationGroupCreateTime: datetime = Field(default=None, frozen=True)
"""
The date and time when the cluster was created.
"""
DataTiering: Literal["enabled", "disabled"] = Field(default=None, frozen=True)
"""
Enables data tiering.
Data tiering is only supported for replication groups using the r6gd node type. This
parameter must be set to true when using r6gd nodes. For more information, see
`Data tiering <https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/data-tiering.html>`_.
"""
AutoMinorVersionUpgrade: "bool | None" = None
"""
If you are running Valkey 7.2 and above, or Redis OSS engine version 6.0 and above,
set this parameter to yes if you want to opt-in to the next auto minor version
upgrade campaign.
This parameter is disabled for previous versions.
"""
NetworkType: "Literal['ipv4', 'ipv6', 'dual_stack'] | None" = None
"""
Must be either ``ipv4`` | ``ipv6`` | ``dual_stack``.
IPv6 is supported for workloads using Valkey 7.2 and above, Redis OSS engine version
6.2 to 7.1 or Memcached engine version 1.6.6 and above on all instances built on the
`Nitro system <http://aws.amazon.com/ec2/nitro/>`_.
"""
IpDiscovery: "Literal['ipv4', 'ipv6'] | None" = None
"""
The network type you choose when modifying a cluster, either ``ipv4`` | ``ipv6``.
IPv6 is supported for workloads using Valkey 7.2 and above, Redis OSS engine version
6.2 to 7.1 or Memcached engine version 1.6.6 and above on all instances built on the
`Nitro system <http://aws.amazon.com/ec2/nitro/>`_.
"""
TransitEncryptionMode: "Literal['preferred', 'required'] | None" = None
"""
A setting that allows you to migrate your clients to use in-transit encryption, with
no downtime.
"""
ClusterMode: "Literal['enabled', 'disabled', 'compatible'] | None" = None
"""
Enabled or Disabled.
To modify cluster mode from Disabled to Enabled, you must first set the cluster mode
to Compatible. Compatible mode allows your Valkey or Redis OSS clients to connect
using both cluster mode enabled and cluster mode disabled. After you migrate all
Valkey or Redis OSS clients to use cluster mode enabled, you can then complete
cluster mode configuration and set the cluster mode to Enabled.
"""
Engine: "str | None" = None
"""
The engine used in a replication group.
The options are valkey, memcached or redis.
"""
@property
def pk(self) -> str | None:
"""
Return the primary key of the model. This is the value of the :py:attr:`ARN`
attribute.
Returns:
The primary key of the model instance.
"""
return self.ARN
@property
def arn(self) -> str | None:
"""
Return the ARN of the model. This is the value of the :py:attr:`ARN`
attribute.
Returns:
The ARN of the model instance.
"""
return self.ARN
@property
def name(self) -> str | None:
"""
Return the name of the model. This is the value of the
:py:attr:`ReplicationGroupId` attribute.
Returns:
The name of the model instance.
"""
return self.ReplicationGroupId
def __hash__(self) -> int:
"""
Return the hash of the model.
This is the value of the
:py:attr:`ARN` attribute.
"""
return hash(self.ARN)
[docs]class CacheParameter(ReadonlyPrimaryBoto3Model):
"""
Describes an individual setting that controls some aspect of ElastiCache behavior.
"""
manager_class: ClassVar[type[Boto3ModelManager]] = CacheParameterManager
ParameterName: "str | None" = None
"""
The name of the parameter.
"""
ParameterValue: "str | None" = None
"""
The value of the parameter.
"""
Description: "str | None" = None
"""
A description of the parameter.
"""
Source: "str | None" = None
"""
The source of the parameter.
"""
DataType: "str | None" = None
"""
The valid data type for the parameter.
"""
AllowedValues: "str | None" = None
"""
The valid range of values for the parameter.
"""
IsModifiable: "bool | None" = None
"""
Indicates whether (``true``) or not (``false``) the parameter can be modified.
Some parameters have security or operational implications that prevent them from
being changed.
"""
MinimumEngineVersion: "str | None" = None
"""
The earliest cache engine version to which the parameter can apply.
"""
ChangeType: "Literal['immediate', 'requires-reboot'] | None" = None
"""
Indicates whether a change to the parameter is applied immediately or requires a
reboot for the change to be applied.
You can force a reboot or wait until the next maintenance window's reboot. For more
information, see
`Rebooting a Cluster <https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/Clusters.Rebooting.html>`_.
"""
@property
def pk(self) -> str | None:
"""
Return the primary key of the model. This is the value of the
:py:attr:`ParameterName` attribute.
Returns:
The primary key of the model instance.
"""
return self.ParameterName
@property
def name(self) -> str | None:
"""
Return the name of the model. This is the value of the
:py:attr:`ParameterName` attribute.
Returns:
The name of the model instance.
"""
return self.ParameterName
def __hash__(self) -> int:
"""
Return the hash of the model.
This is the value of the
:py:attr:`ParameterName` attribute.
"""
return hash(self.ParameterName)
[docs]class ElastiCacheAuthentication(Boto3Model):
"""
Indicates whether the user requires a password to authenticate.
"""
Type: "Literal['password', 'no-password', 'iam'] | None" = None
"""
Indicates whether the user requires a password to authenticate.
"""
PasswordCount: "int | None" = None
"""
The number of passwords belonging to the user.
The maximum is two.
"""
[docs]class ElastiCacheUser(TagsDictMixin, PrimaryBoto3Model):
tag_class: ClassVar[type[Boto3Model]] = Tag
manager_class: ClassVar[type[Boto3ModelManager]] = ElastiCacheUserManager
Status: "str | None" = "active"
"""
Indicates the user status.
Can be "active", "modifying" or "deleting".
"""
UserId: "str | None" = None
"""
The ID of the user.
"""
UserName: "str | None" = None
"""
The username of the user.
"""
Engine: "str | None" = None
"""
The options are valkey or redis.
"""
MinimumEngineVersion: str = Field(default=None, frozen=True)
"""
The minimum engine version required, which is Redis OSS 6.0.
"""
AccessString: "str | None" = None
"""
Access permissions string used for this user.
"""
UserGroupIds: "builtins.list[str]" = Field(default_factory=list, frozen=True)
"""
Returns a list of the user group IDs the user belongs to.
"""
Authentication: ElastiCacheAuthentication = Field(default=None, frozen=True)
"""
Denotes whether the user requires a password to authenticate.
"""
ARN: str = Field(default=None, frozen=True)
"""
The Amazon Resource Name (ARN) of the user.
"""
Tags: "builtins.list[Tag] | None" = Field(default_factory=list)
"""
The tags associated with the user.
"""
@property
def pk(self) -> str | None:
"""
Return the primary key of the model. This is the value of the :py:attr:`ARN`
attribute.
Returns:
The primary key of the model instance.
"""
return self.ARN
@property
def arn(self) -> str | None:
"""
Return the ARN of the model. This is the value of the :py:attr:`ARN`
attribute.
Returns:
The ARN of the model instance.
"""
return self.ARN
@property
def name(self) -> str | None:
"""
Return the name of the model. This is the value of the :py:attr:`UserId`
attribute.
Returns:
The name of the model instance.
"""
return self.UserId
def __hash__(self) -> int:
"""
Return the hash of the model.
This is the value of the
:py:attr:`ARN` attribute.
"""
return hash(self.ARN)
[docs]class UserGroupPendingChanges(Boto3Model):
"""
Returns the updates being applied to the user group.
"""
UserIdsToRemove: "builtins.list[str] | None" = Field(default_factory=list)
"""
The list of user IDs to remove.
"""
UserIdsToAdd: "builtins.list[str] | None" = Field(default_factory=list)
"""
The list of user IDs to add.
"""
[docs]class ElastiCacheUserGroup(TagsDictMixin, PrimaryBoto3Model):
tag_class: ClassVar[type[Boto3Model]] = Tag
manager_class: ClassVar[type[Boto3ModelManager]] = ElastiCacheUserGroupManager
Status: "str | None" = "active"
"""
Indicates user group status.
Can be "creating", "active", "modifying", "deleting".
"""
UserGroupId: "str | None" = None
"""
The ID of the user group.
"""
Engine: "str | None" = None
"""
The options are valkey or redis.
"""
UserIds: "builtins.list[str] | None" = Field(default_factory=list)
"""
The list of user IDs that belong to the user group.
"""
MinimumEngineVersion: str = Field(default=None, frozen=True)
"""
The minimum engine version required, which is Redis OSS 6.0.
"""
PendingChanges: UserGroupPendingChanges = Field(default=None, frozen=True)
"""
A list of updates being applied to the user group.
"""
ReplicationGroups: "builtins.list[str]" = Field(default_factory=list, frozen=True)
"""
A list of replication groups that the user group can access.
"""
ServerlessCaches: "builtins.list[str]" = Field(default_factory=list, frozen=True)
"""
Indicates which serverless caches the specified user group is associated with.
Available for Valkey, Redis OSS and Serverless Memcached only.
"""
ARN: str = Field(default=None, frozen=True)
"""
The Amazon Resource Name (ARN) of the user group.
"""
Tags: "builtins.list[Tag] | None" = Field(default_factory=list)
"""
The tags associated with the user group.
"""
@property
def pk(self) -> str | None:
"""
Return the primary key of the model. This is the value of the :py:attr:`ARN`
attribute.
Returns:
The primary key of the model instance.
"""
return self.ARN
@property
def arn(self) -> str | None:
"""
Return the ARN of the model. This is the value of the :py:attr:`ARN`
attribute.
Returns:
The ARN of the model instance.
"""
return self.ARN
@property
def name(self) -> str | None:
"""
Return the name of the model. This is the value of the :py:attr:`UserGroupId`
attribute.
Returns:
The name of the model instance.
"""
return self.UserGroupId
def __hash__(self) -> int:
"""
Return the hash of the model.
This is the value of the
:py:attr:`ARN` attribute.
"""
return hash(self.ARN)
# =======================
# Request/Response Models
# =======================
[docs]class LogDeliveryConfigurationRequest(Boto3Model):
"""
Specifies the destination, format and type of the logs.
"""
LogType: "Literal['slow-log', 'engine-log'] | None" = None
"""
Refers to `slow-log <https://redis.io/commands/slowlog>`_ or engine-log..
"""
DestinationType: "Literal['cloudwatch-logs', 'kinesis-firehose'] | None" = None
"""
Specify either ``cloudwatch-logs`` or ``kinesis-firehose`` as the destination type.
"""
DestinationDetails: "ElastiCacheDestinationDetails | None" = None
"""
Configuration details of either a CloudWatch Logs destination or Kinesis Data
Firehose destination.
"""
LogFormat: "Literal['text', 'json'] | None" = None
"""
Specifies either JSON or TEXT.
"""
Enabled: "bool | None" = None
"""
Specify if log delivery is enabled.
Default ``true``.
"""
[docs]class CreateCacheClusterResult(Boto3Model):
ElasticacheCluster: CacheCluster = Field(default=None, alias="CacheCluster")
"""
Contains all of the attributes of a specific cluster.
"""
[docs]class ModifyCacheClusterResult(Boto3Model):
ElasticacheCluster: CacheCluster = Field(default=None, alias="CacheCluster")
"""
Contains all of the attributes of a specific cluster.
"""
[docs]class DeleteCacheClusterResult(Boto3Model):
ElasticacheCluster: CacheCluster = Field(default=None, alias="CacheCluster")
"""
Contains all of the attributes of a specific cluster.
"""
[docs]class CacheClusterMessage(Boto3Model):
"""
Represents the output of a ``DescribeCacheClusters`` operation.
"""
Marker: "str | None" = None
"""
Provides an identifier to allow retrieval of paginated results.
"""
CacheClusters: "builtins.list[CacheCluster] | None" = Field(default_factory=list)
"""
A list of clusters.
Each item in the list contains detailed information about one cluster.
"""
[docs]class CreateCacheParameterGroupResult(Boto3Model):
ElasticacheParameterGroup: CacheParameterGroup = Field(
default=None, alias="CacheParameterGroup"
)
"""
Represents the output of a ``CreateCacheParameterGroup`` operation.
"""
[docs]class CacheParameterGroupNameMessage(Boto3Model):
"""Represents the output of one of the following operations:
* ``ModifyCacheParameterGroup``
* ``ResetCacheParameterGroup``
"""
CacheParameterGroupName: "str | None" = None
"""
The name of the cache parameter group.
"""
[docs]class CacheParameterGroupsMessage(Boto3Model):
"""
Represents the output of a ``DescribeCacheParameterGroups`` operation.
"""
Marker: "str | None" = None
"""
Provides an identifier to allow retrieval of paginated results.
"""
CacheParameterGroups: "builtins.list[CacheParameterGroup] | None" = Field(
default_factory=list
)
"""
A list of cache parameter groups.
Each element in the list contains detailed information about one cache parameter
group.
"""
[docs]class CacheNodeTypeSpecificValue(Boto3Model):
"""
A value that applies only to a certain cache node type.
"""
CacheNodeType: "str | None" = None
"""
The cache node type for which this value applies.
"""
Value: "str | None" = None
"""
The value for the cache node type.
"""
[docs]class CacheNodeTypeSpecificParameter(Boto3Model):
"""
A parameter that has a different value for each cache node type it is applied to.
For example, in a Valkey or Redis OSS
cluster, a ``cache.m1.large`` cache node type would have a larger ``maxmemory`` value than a ``cache.m1.small`` type.
"""
ParameterName: "str | None" = None
"""
The name of the parameter.
"""
Description: "str | None" = None
"""
A description of the parameter.
"""
Source: "str | None" = None
"""
The source of the parameter value.
"""
DataType: "str | None" = None
"""
The valid data type for the parameter.
"""
AllowedValues: "str | None" = None
"""
The valid range of values for the parameter.
"""
IsModifiable: "bool | None" = None
"""
Indicates whether (``true``) or not (``false``) the parameter can be modified.
Some parameters have security or operational implications that prevent them from
being changed.
"""
MinimumEngineVersion: "str | None" = None
"""
The earliest cache engine version to which the parameter can apply.
"""
CacheNodeTypeSpecificValues: "builtins.list[CacheNodeTypeSpecificValue] | None" = (
Field(default_factory=list)
)
"""
A list of cache node types and their corresponding values for this parameter.
"""
ChangeType: "Literal['immediate', 'requires-reboot'] | None" = None
"""
Indicates whether a change to the parameter is applied immediately or requires a
reboot for the change to be applied.
You can force a reboot or wait until the next maintenance window's reboot. For more
information, see
`Rebooting a Cluster <https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/Clusters.Rebooting.html>`_.
"""
[docs]class CacheParameterGroupDetails(Boto3Model):
"""
Represents the output of a ``DescribeCacheParameters`` operation.
"""
Marker: "str | None" = None
"""
Provides an identifier to allow retrieval of paginated results.
"""
Parameters: "builtins.list[CacheParameter] | None" = Field(default_factory=list)
"""
A list of Parameter instances.
"""
CacheNodeTypeSpecificParameters: "builtins.list[CacheNodeTypeSpecificParameter] | None" = Field(
default_factory=list
)
"""
A list of parameters specific to a particular cache node type.
Each element in the list contains detailed information about one parameter.
"""
[docs]class CreateCacheSubnetGroupResult(Boto3Model):
ElasticacheSubnetGroup: CacheSubnetGroup = Field(
default=None, alias="CacheSubnetGroup"
)
"""
Represents the output of one of the following operations:
"""
[docs]class ModifyCacheSubnetGroupResult(Boto3Model):
ElasticacheSubnetGroup: CacheSubnetGroup = Field(
default=None, alias="CacheSubnetGroup"
)
"""
Represents the output of one of the following operations:
"""
[docs]class CacheSubnetGroupMessage(Boto3Model):
"""
Represents the output of a ``DescribeCacheSubnetGroups`` operation.
"""
Marker: "str | None" = None
"""
Provides an identifier to allow retrieval of paginated results.
"""
CacheSubnetGroups: "builtins.list[CacheSubnetGroup] | None" = Field(
default_factory=list
)
"""
A list of cache subnet groups.
Each element in the list contains detailed information about one group.
"""
[docs]class CreateCacheSecurityGroupResult(Boto3Model):
ElasticacheSecurityGroup: CacheSecurityGroup = Field(
default=None, alias="CacheSecurityGroup"
)
"""
Represents the output of one of the following operations:
"""
[docs]class CacheSecurityGroupMessage(Boto3Model):
"""
Represents the output of a ``DescribeCacheSecurityGroups`` operation.
"""
Marker: "str | None" = None
"""
Provides an identifier to allow retrieval of paginated results.
"""
CacheSecurityGroups: "builtins.list[CacheSecurityGroup] | None" = Field(
default_factory=list
)
"""
A list of cache security groups.
Each element in the list contains detailed information about one group.
"""
[docs]class AuthorizeCacheSecurityGroupIngressResult(Boto3Model):
ElasticacheSecurityGroup: CacheSecurityGroup = Field(
default=None, alias="CacheSecurityGroup"
)
"""
Represents the output of one of the following operations:
"""
[docs]class RevokeCacheSecurityGroupIngressResult(Boto3Model):
ElasticacheSecurityGroup: CacheSecurityGroup = Field(
default=None, alias="CacheSecurityGroup"
)
"""
Represents the output of one of the following operations:
"""
[docs]class NodeGroupConfiguration(Boto3Model):
"""
Node group (shard) configuration options.
Each node group (shard) configuration has the following: ``Slots``,
``PrimaryAvailabilityZone``, ``ReplicaAvailabilityZones``, ``ReplicaCount``.
"""
NodeGroupId: "str | None" = None
"""
Either the ElastiCache supplied 4-digit id or a user supplied id for the node group
these configuration values apply to.
"""
Slots: "str | None" = None
"""
A string that specifies the keyspace for a particular node group.
Keyspaces range from 0 to 16,383. The string is in the
format ``startkey-endkey``.
"""
ReplicaCount: "int | None" = None
"""
The number of read replica nodes in this node group (shard).
"""
PrimaryAvailabilityZone: "str | None" = None
"""
The Availability Zone where the primary node of this node group (shard) is launched.
"""
ReplicaAvailabilityZones: "builtins.list[str] | None" = Field(default_factory=list)
"""
A list of Availability Zones to be used for the read replicas.
The number of Availability Zones in this list must match
the value of ``ReplicaCount`` or ``ReplicasPerNodeGroup`` if not specified.
"""
PrimaryOutpostArn: "str | None" = None
"""
The outpost ARN of the primary node.
"""
ReplicaOutpostArns: "builtins.list[str] | None" = Field(default_factory=list)
"""
The outpost ARN of the node replicas.
"""
[docs]class CreateReplicationGroupResult(Boto3Model):
ElasticacheReplicationGroup: ReplicationGroup = Field(
default=None, alias="ReplicationGroup"
)
"""
Contains all of the attributes of a specific Valkey or Redis OSS replication group.
"""
[docs]class ModifyReplicationGroupResult(Boto3Model):
ElasticacheReplicationGroup: ReplicationGroup = Field(
default=None, alias="ReplicationGroup"
)
"""
Contains all of the attributes of a specific Valkey or Redis OSS replication group.
"""
[docs]class DeleteReplicationGroupResult(Boto3Model):
ElasticacheReplicationGroup: ReplicationGroup = Field(
default=None, alias="ReplicationGroup"
)
"""
Contains all of the attributes of a specific Valkey or Redis OSS replication group.
"""
[docs]class ReplicationGroupMessage(Boto3Model):
"""
Represents the output of a ``DescribeReplicationGroups`` operation.
"""
Marker: "str | None" = None
"""
Provides an identifier to allow retrieval of paginated results.
"""
ReplicationGroups: "builtins.list[ReplicationGroup] | None" = Field(
default_factory=list
)
"""
A list of replication groups.
Each item in the list contains detailed information about one replication group.
"""
[docs]class AuthenticationMode(Boto3Model):
"""
Specifies the authentication mode to use.
"""
Type: "Literal['password', 'no-password-required', 'iam'] | None" = None
"""
Specifies the authentication type.
Possible options are IAM authentication, password and no password.
"""
Passwords: "builtins.list[str] | None" = Field(default_factory=list)
"""
Specifies the passwords to use for authentication if ``Type`` is set to
``password``.
"""
[docs]class DescribeUsersResult(Boto3Model):
Users: "builtins.list[ElastiCacheUser] | None" = Field(default_factory=list)
"""
A list of users.
"""
Marker: "str | None" = None
"""
An optional marker returned from a prior request.
Use this marker for pagination of results from this operation. If this parameter is
specified, the response includes only records beyond the marker, up to the value
specified by MaxRecords. >
"""
[docs]class DescribeUserGroupsResult(Boto3Model):
UserGroups: "builtins.list[ElastiCacheUserGroup] | None" = Field(
default_factory=list
)
"""
Returns a list of user groups.
"""
Marker: "str | None" = None
"""
An optional marker returned from a prior request.
Use this marker for pagination of results from this operation. If this parameter is
specified, the response includes only records beyond the marker, up to the value
specified by MaxRecords.>
"""