Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.
# Python et ElastiCache
Dans ce didacticiel, vous allez utiliser le AWS SDK pour Python (Boto3) pour écrire des programmes simples permettant d'effectuer les opérations suivantes : ElastiCache
+ Créer ElastiCache pour les clusters Redis OSS (mode cluster activé et mode cluster désactivé)
+ Vérifiez si des utilisateurs ou des groupes d'utilisateurs existent, sinon créez-les. (Cette fonctionnalité est disponible avec Valkey 7.2 et versions ultérieures, et avec Redis OSS 6.0 à 7.1.)
+ Connect à ElastiCache
+ Effectuez des opérations telles que le réglage et l'obtention de chaînes, la lecture et l'écriture sur des streams, ainsi que la publication et l'abonnement à Pub/Sub une chaîne.
Au cours de ce didacticiel, vous pouvez consulter la documentation du AWS SDK pour Python (Boto). La section suivante est spécifique à ElastiCache : client de [ElastiCache bas niveau](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/elasticache.html)
## Prérequis du didacticiel
+ Configurez une cléAWS d'accès pour utiliser le AWS SDKs. Pour de plus amples informations, veuillez consulter [Con ElastiCachefiguration](set-up.md).
+ Installer Python 3.0 ou version ultérieure. Pour plus d'informations, rendez-vous sur [https://www.python.org/downloads](https://www.python.org/downloads). Pour plus d'informations, consultez la page [Quickstart](https://boto3.amazonaws.com/v1/documentation/api/latest/guide/quickstart.html) dans la documentation Boto 3.
**Topics**
+ [Prérequis du didacticiel](#ElastiCache-Getting-Started-Tutorials-Prerquisites)
+ [Tutoriel : Création de ElastiCache clusters et d'utilisateurs](#ElastiCache-Getting-Started-Tutorials-Create-Cluster-and-Users)
+ [Tutoriel : Connexion à ElastiCache](#ElastiCache-Getting-Started-Tutorials-Connecting)
+ [Exemples d’utilisation](#ElastiCache-Getting-Started-Tutorials-Usage)
## Tutoriel : Création de ElastiCache clusters et d'utilisateurs
Les exemples suivants utilisent le SDK boto3 ElastiCache pour les opérations de gestion Redis OSS (création de clusters ou d'utilisateurs) et redis-py-cluster redis-py/ pour le traitement des données.
**Topics**
+ [Créer un cluster en mode cluster désactivé](#ElastiCache-Getting-Started-Tutorials-Create-Cluster)
+ [Créer un cluster en mode cluster désactivé avec TLS et RBAC](#ElastiCache-Getting-Started-Tutorials-RBAC)
+ [Créer un cluster Redis en mode activé](#ElastiCache-Getting-Started-Tutorials-Cluster-Enabled)
+ [Créer un cluster en mode cluster activé avec TLS et RBAC](#ElastiCache-Getting-Started-Tutorials-Cluster-RBAC)
+ [Vérifiez s'ils users/usergroup existent, sinon créez-les](#ElastiCache-Getting-Started-Tutorials-Users)
### Créer un cluster en mode cluster désactivé
Copiez le programme suivant et collez-le dans un fichier nommé *CreateClusterModeDisabledCluster.py.*
```
import boto3
import logging
logging.basicConfig(level=logging.INFO)
client = boto3.client('elasticache')
def create_cluster_mode_disabled(CacheNodeType='cache.t3.small',EngineVersion='6.0',NumCacheClusters=2,ReplicationGroupDescription='Sample cluster',ReplicationGroupId=None):
"""Creates an ElastiCache Cluster with cluster mode disabled
Returns a dictionary with the API response
:param CacheNodeType: Node type used on the cluster. If not specified, cache.t3.small will be used
Refer to https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/CacheNodes.SupportedTypes.html for supported node types
:param EngineVersion: Engine version to be used. If not specified, latest will be used.
:param NumCacheClusters: Number of nodes in the cluster. Minimum 1 (just a primary node) and maximun 6 (1 primary and 5 replicas).
If not specified, cluster will be created with 1 primary and 1 replica.
:param ReplicationGroupDescription: Description for the cluster.
:param ReplicationGroupId: Name for the cluster
:return: dictionary with the API results
"""
if not ReplicationGroupId:
return 'ReplicationGroupId parameter is required'
response = client.create_replication_group(
AutomaticFailoverEnabled=True,
CacheNodeType=CacheNodeType,
Engine='valkey',
EngineVersion=EngineVersion,
NumCacheClusters=NumCacheClusters,
ReplicationGroupDescription=ReplicationGroupDescription,
ReplicationGroupId=ReplicationGroupId,
SnapshotRetentionLimit=30,
)
return response
if __name__ == '__main__':
# Creates an ElastiCache Cluster mode disabled cluster, based on cache.m6g.large nodes, Valkey 8.0, one primary and two replicas
elasticacheResponse = create_cluster_mode_disabled(
#CacheNodeType='cache.m6g.large',
EngineVersion='8.0',
NumCacheClusters=3,
ReplicationGroupDescription='Valkey cluster mode disabled with replicas',
ReplicationGroupId='valkey202104053'
)
logging.info(elasticacheResponse)
```
Pour exécuter le programme, saisissez la commande suivante.
`python CreateClusterModeDisabledCluster.py`
Pour de plus amples informations, veuillez consulter [Gestion des clusters dans ElastiCache](Clusters.md).
### Créer un cluster en mode cluster désactivé avec TLS et RBAC
Pour garantir la sécurité, vous pouvez utiliser TLS (Transport Layer Security) et RBAC (Role Based Access Control) lors de la création d'un cluster en mode cluster désactivé. Contrairement à Valkey ou Redis OSS AUTH, où tous les clients authentifiés ont un accès complet au groupe de réplication si leur jeton est authentifié, le RBAC vous permet de contrôler l'accès au cluster par le biais de groupes d'utilisateurs. Ces groupes d'utilisateurs sont conçus comme un moyen d'organiser l'accès aux groupes de réplication. Pour de plus amples informations, veuillez consulter [Contrôle d'accès basé sur les rôles (RBAC)](Clusters.RBAC.md).
Copiez le programme suivant et collez-le dans un fichier nommé *ClusterModeDisabledWithRBAC.py*.
```
import boto3
import logging
logging.basicConfig(level=logging.INFO)
client = boto3.client('elasticache')
def create_cluster_mode_disabled_rbac(CacheNodeType='cache.t3.small',EngineVersion='6.0',NumCacheClusters=2,ReplicationGroupDescription='Sample cluster',ReplicationGroupId=None, UserGroupIds=None, SecurityGroupIds=None,CacheSubnetGroupName=None):
"""Creates an ElastiCache Cluster with cluster mode disabled and RBAC
Returns a dictionary with the API response
:param CacheNodeType: Node type used on the cluster. If not specified, cache.t3.small will be used
Refer to https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/CacheNodes.SupportedTypes.html for supported node types
:param EngineVersion: Engine version to be used. If not specified, latest will be used.
:param NumCacheClusters: Number of nodes in the cluster. Minimum 1 (just a primary node) and maximun 6 (1 primary and 5 replicas).
If not specified, cluster will be created with 1 primary and 1 replica.
:param ReplicationGroupDescription: Description for the cluster.
:param ReplicationGroupId: Mandatory name for the cluster.
:param UserGroupIds: The ID of the user group to be assigned to the cluster.
:param SecurityGroupIds: List of security groups to be assigned. If not defined, default will be used
:param CacheSubnetGroupName: subnet group where the cluster will be placed. If not defined, default will be used.
:return: dictionary with the API results
"""
if not ReplicationGroupId:
return {'Error': 'ReplicationGroupId parameter is required'}
elif not isinstance(UserGroupIds,(list)):
return {'Error': 'UserGroupIds parameter is required and must be a list'}
params={'AutomaticFailoverEnabled': True,
'CacheNodeType': CacheNodeType,
'Engine': 'valkey',
'EngineVersion': EngineVersion,
'NumCacheClusters': NumCacheClusters,
'ReplicationGroupDescription': ReplicationGroupDescription,
'ReplicationGroupId': ReplicationGroupId,
'SnapshotRetentionLimit': 30,
'TransitEncryptionEnabled': True,
'UserGroupIds':UserGroupIds
}
# defaults will be used if CacheSubnetGroupName or SecurityGroups are not explicit.
if isinstance(SecurityGroupIds,(list)):
params.update({'SecurityGroupIds':SecurityGroupIds})
if CacheSubnetGroupName:
params.update({'CacheSubnetGroupName':CacheSubnetGroupName})
response = client.create_replication_group(**params)
return response
if __name__ == '__main__':
# Creates an ElastiCache Cluster mode disabled cluster, based on cache.m6g.large nodes, Valkey 8.0, one primary and two replicas.
# Assigns the existent user group "mygroup" for RBAC authentication
response=create_cluster_mode_disabled_rbac(
CacheNodeType='cache.m6g.large',
EngineVersion='8.0',
NumCacheClusters=3,
ReplicationGroupDescription='Valkey cluster mode disabled with replicas',
ReplicationGroupId='valkey202104',
UserGroupIds=[
'mygroup'
],
SecurityGroupIds=[
'sg-7cc73803'
],
CacheSubnetGroupName='default'
)
logging.info(response)
```
Pour exécuter le programme, saisissez la commande suivante.
`python ClusterModeDisabledWithRBAC.py`
Pour de plus amples informations, veuillez consulter [Gestion des clusters dans ElastiCache](Clusters.md).
### Créer un cluster Redis en mode activé
Copiez le programme suivant et collez-le dans un fichier nommé *ClusterModeEnabled.py.*
```
import boto3
import logging
logging.basicConfig(level=logging.INFO)
client = boto3.client('elasticache')
def create_cluster_mode_enabled(CacheNodeType='cache.t3.small',EngineVersion='6.0',NumNodeGroups=1,ReplicasPerNodeGroup=1, ReplicationGroupDescription='Sample cache with cluster mode enabled',ReplicationGroupId=None):
"""Creates an ElastiCache Cluster with cluster mode enabled
Returns a dictionary with the API response
:param CacheNodeType: Node type used on the cluster. If not specified, cache.t3.small will be used
Refer to https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/CacheNodes.SupportedTypes.html for supported node types
:param EngineVersion: Engine version to be used. If not specified, latest will be used.
:param NumNodeGroups: Number of shards in the cluster. Minimum 1 and maximun 90.
If not specified, cluster will be created with 1 shard.
:param ReplicasPerNodeGroup: Number of replicas per shard. If not specified 1 replica per shard will be created.
:param ReplicationGroupDescription: Description for the cluster.
:param ReplicationGroupId: Name for the cluster
:return: dictionary with the API results
"""
if not ReplicationGroupId:
return 'ReplicationGroupId parameter is required'
response = client.create_replication_group(
AutomaticFailoverEnabled=True,
CacheNodeType=CacheNodeType,
Engine='valkey',
EngineVersion=EngineVersion,
ReplicationGroupDescription=ReplicationGroupDescription,
ReplicationGroupId=ReplicationGroupId,
# Creates a cluster mode enabled cluster with 1 shard(NumNodeGroups), 1 primary node (implicit) and 2 replicas (replicasPerNodeGroup)
NumNodeGroups=NumNodeGroups,
ReplicasPerNodeGroup=ReplicasPerNodeGroup,
CacheParameterGroupName='default.valkey7.2.cluster.on'
)
return response
# Creates a cluster mode enabled
response = create_cluster_mode_enabled(
CacheNodeType='cache.m6g.large',
EngineVersion='6.0',
ReplicationGroupDescription='Valkey cluster mode enabled with replicas',
ReplicationGroupId='valkey20210',
# Creates a cluster mode enabled cluster with 1 shard(NumNodeGroups), 1 primary (implicit) and 2 replicas (replicasPerNodeGroup)
NumNodeGroups=2,
ReplicasPerNodeGroup=1,
)
logging.info(response)
```
Pour exécuter le programme, saisissez la commande suivante.
`python ClusterModeEnabled.py`
Pour de plus amples informations, veuillez consulter [Gestion des clusters dans ElastiCache](Clusters.md).
### Créer un cluster en mode cluster activé avec TLS et RBAC
Pour garantir la sécurité, vous pouvez utiliser TLS (Transport Layer Security) et RBAC (Role Based Access Control) lors de la création d'un cluster en mode cluster activé. Contrairement à Valkey ou Redis OSS AUTH, où tous les clients authentifiés ont un accès complet au groupe de réplication si leur jeton est authentifié, le RBAC vous permet de contrôler l'accès au cluster par le biais de groupes d'utilisateurs. Ces groupes d'utilisateurs sont conçus comme un moyen d'organiser l'accès aux groupes de réplication. Pour de plus amples informations, veuillez consulter [Contrôle d'accès basé sur les rôles (RBAC)](Clusters.RBAC.md).
Copiez le programme suivant et collez-le dans un fichier nommé *ClusterModeEnabledWithRBAC.py*.
```
import boto3
import logging
logging.basicConfig(level=logging.INFO)
client = boto3.client('elasticache')
def create_cluster_mode_enabled(CacheNodeType='cache.t3.small',EngineVersion='6.0',NumNodeGroups=1,ReplicasPerNodeGroup=1, ReplicationGroupDescription='Sample cache with cluster mode enabled',ReplicationGroupId=None,UserGroupIds=None, SecurityGroupIds=None,CacheSubnetGroupName=None,CacheParameterGroupName='default.valkey7.2.cluster.on'):
"""Creates an ElastiCache Cluster with cluster mode enabled and RBAC
Returns a dictionary with the API response
:param CacheNodeType: Node type used on the cluster. If not specified, cache.t3.small will be used
Refer to https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/CacheNodes.SupportedTypes.html for supported node types
:param EngineVersion: Engine version to be used. If not specified, latest will be used.
:param NumNodeGroups: Number of shards in the cluster. Minimum 1 and maximun 90.
If not specified, cluster will be created with 1 shard.
:param ReplicasPerNodeGroup: Number of replicas per shard. If not specified 1 replica per shard will be created.
:param ReplicationGroupDescription: Description for the cluster.
:param ReplicationGroupId: Name for the cluster.
:param CacheParameterGroupName: Parameter group to be used. Must be compatible with the engine version and cluster mode enabled.
:return: dictionary with the API results
"""
if not ReplicationGroupId:
return 'ReplicationGroupId parameter is required'
elif not isinstance(UserGroupIds,(list)):
return {'Error': 'UserGroupIds parameter is required and must be a list'}
params={'AutomaticFailoverEnabled': True,
'CacheNodeType': CacheNodeType,
'Engine': 'valkey',
'EngineVersion': EngineVersion,
'ReplicationGroupDescription': ReplicationGroupDescription,
'ReplicationGroupId': ReplicationGroupId,
'SnapshotRetentionLimit': 30,
'TransitEncryptionEnabled': True,
'UserGroupIds':UserGroupIds,
'NumNodeGroups': NumNodeGroups,
'ReplicasPerNodeGroup': ReplicasPerNodeGroup,
'CacheParameterGroupName': CacheParameterGroupName
}
# defaults will be used if CacheSubnetGroupName or SecurityGroups are not explicit.
if isinstance(SecurityGroupIds,(list)):
params.update({'SecurityGroupIds':SecurityGroupIds})
if CacheSubnetGroupName:
params.update({'CacheSubnetGroupName':CacheSubnetGroupName})
response = client.create_replication_group(**params)
return response
if __name__ == '__main__':
# Creates a cluster mode enabled cluster
response = create_cluster_mode_enabled(
CacheNodeType='cache.m6g.large',
EngineVersion='7.2',
ReplicationGroupDescription='Valkey cluster mode enabled with replicas',
ReplicationGroupId='valkey2021',
# Creates a cluster mode enabled cluster with 1 shard(NumNodeGroups), 1 primary (implicit) and 2 replicas (replicasPerNodeGroup)
NumNodeGroups=2,
ReplicasPerNodeGroup=1,
UserGroupIds=[
'mygroup'
],
SecurityGroupIds=[
'sg-7cc73803'
],
CacheSubnetGroupName='default'
)
logging.info(response)
```
Pour exécuter le programme, saisissez la commande suivante.
`python ClusterModeEnabledWithRBAC.py`
Pour de plus amples informations, veuillez consulter [Gestion des clusters dans ElastiCache](Clusters.md).
### Vérifiez s'ils users/usergroup existent, sinon créez-les
Avec RBAC, vous créez des utilisateurs et leur attribuez des autorisations spécifiques à l'aide d'une chaîne d'accès. Vous assignez les utilisateurs à des groupes d'utilisateurs correspondant à un rôle spécifique (administrateurs, ressources humaines) qui sont ensuite déployés dans un ou plusieurs groupes ElastiCache de réplication Redis OSS. Vous pouvez ainsi établir des limites de sécurité entre les clients utilisant le ou les mêmes groupes de réplication Valkey ou Redis OSS et empêcher les clients d'accéder aux données des autres. Pour de plus amples informations, veuillez consulter [Contrôle d'accès basé sur les rôles (RBAC)](Clusters.RBAC.md).
Copiez le programme suivant et collez-le dans un fichier nommé *UserAndUserGroups.py.* Mettez à jour le mécanisme de fourniture des informations d'identification. Dans cet exemple, les informations d'identification sont affichées comme remplaçables et un élément non déclaré leur est affecté. Évitez de coder en dur les informations d'identification.
Cet exemple utilise une chaîne d'accès contenant les autorisations de l'utilisateur. Pour plus d'informations sur les chaînes d'accès, voir[Définition des autorisations à l'aide d'une chaîne d'accès](Clusters.RBAC.md#Access-string).
```
import boto3
import logging
logging.basicConfig(level=logging.INFO)
client = boto3.client('elasticache')
def check_user_exists(UserId):
"""Checks if UserId exists
Returns True if UserId exists, otherwise False
:param UserId: ElastiCache User ID
:return: True|False
"""
try:
response = client.describe_users(
UserId=UserId,
)
if response['Users'][0]['UserId'].lower() == UserId.lower():
return True
except Exception as e:
if e.response['Error']['Code'] == 'UserNotFound':
logging.info(e.response['Error'])
return False
else:
raise
def check_group_exists(UserGroupId):
"""Checks if UserGroupID exists
Returns True if Group ID exists, otherwise False
:param UserGroupId: ElastiCache User ID
:return: True|False
"""
try:
response = client.describe_user_groups(
UserGroupId=UserGroupId
)
if response['UserGroups'][0]['UserGroupId'].lower() == UserGroupId.lower():
return True
except Exception as e:
if e.response['Error']['Code'] == 'UserGroupNotFound':
logging.info(e.response['Error'])
return False
else:
raise
def create_user(UserId=None,UserName=None,Password=None,AccessString=None):
"""Creates a new user
Returns the ARN for the newly created user or the error message
:param UserId: ElastiCache user ID. User IDs must be unique
:param UserName: ElastiCache user name. ElastiCache allows multiple users with the same name as long as the associated user ID is unique.
:param Password: Password for user. Must have at least 16 chars.
:param AccessString: Access string with the permissions for the user.
:return: user ARN
"""
try:
response = client.create_user(
UserId=UserId,
UserName=UserName,
Engine='Redis',
Passwords=[Password],
AccessString=AccessString,
NoPasswordRequired=False
)
return response['ARN']
except Exception as e:
logging.info(e.response['Error'])
return e.response['Error']
def create_group(UserGroupId=None, UserIds=None):
"""Creates a new group.
A default user is required (mandatory) and should be specified in the UserIds list
Return: Group ARN
:param UserIds: List with user IDs to be associated with the new group. A default user is required
:param UserGroupId: The ID (name) for the group
:return: Group ARN
"""
try:
response = client.create_user_group(
UserGroupId=UserGroupId,
Engine='Redis',
UserIds=UserIds
)
return response['ARN']
except Exception as e:
logging.info(e.response['Error'])
if __name__ == '__main__':
groupName='mygroup2'
userName = 'myuser2'
userId=groupName+'-'+userName
# Creates a new user if the user ID does not exist.
for tmpUserId,tmpUserName in [ (userId,userName), (groupName+'-default','default')]:
if not check_user_exists(tmpUserId):
response=create_user(UserId=tmpUserId, UserName=EXAMPLE,Password=EXAMPLE,AccessString='on ~* +@all')
logging.info(response)
# assigns the new user ID to the user group
if not check_group_exists(groupName):
UserIds = [ userId , groupName+'-default']
response=create_group(UserGroupId=groupName,UserIds=UserIds)
logging.info(response)
```
Pour exécuter le programme, saisissez la commande suivante.
`python UserAndUserGroups.py`
## Tutoriel : Connexion à ElastiCache
Les exemples suivants utilisent le client Valkey ou Redis OSS pour se connecter à. ElastiCache
**Topics**
+ [Connexion à un cluster en mode cluster désactivé](#ElastiCache-Getting-Started-Tutorials-Connecting-cluster-mode-disabled)
+ [Connexion à un cluster en mode cluster activé](#ElastiCache-Getting-Started-Tutorials-Connecting-cluster-mode-enabled)
### Connexion à un cluster en mode cluster désactivé
Copiez le programme suivant et collez-le dans un fichier nommé *ConnectClusterModeDisabled.py.* Mettez à jour le mécanisme de fourniture des informations d'identification. Dans cet exemple, les informations d'identification sont affichées comme remplaçables et un élément non déclaré leur est affecté. Évitez de coder en dur les informations d'identification.
```
from redis import Redis
import logging
logging.basicConfig(level=logging.INFO)
redis = Redis(host='primary.xxx.yyyyyy.zzz1.cache.amazonaws.com', port=6379, decode_responses=True, ssl=True, username=example, password=EXAMPLE)
if redis.ping():
logging.info("Connected to Redis")
```
Pour exécuter le programme, saisissez la commande suivante.
`python ConnectClusterModeDisabled.py`
### Connexion à un cluster en mode cluster activé
Copiez le programme suivant et collez-le dans un fichier nommé *ConnectClusterModeEnabled.py.*
```
from rediscluster import RedisCluster
import logging
logging.basicConfig(level=logging.INFO)
redis = RedisCluster(startup_nodes=[{"host": "xxx.yyy.clustercfg.zzz1.cache.amazonaws.com","port": "6379"}], decode_responses=True,skip_full_coverage_check=True)
if redis.ping():
logging.info("Connected to Redis")
```
Pour exécuter le programme, saisissez la commande suivante.
`python ConnectClusterModeEnabled.py`
## Exemples d’utilisation
Les exemples suivants utilisent le SDK boto3 ElastiCache pour fonctionner avec ElastiCache Redis OSS.
**Topics**
+ [Définir et obtenir des chaînes](#ElastiCache-Getting-Started-Tutorials-set-strings)
+ [Définir et obtenir un hachage avec plusieurs éléments](#ElastiCache-Getting-Started-Tutorials-set-hash)
+ [Publier (écrire) et s'abonner (lire) depuis une Pub/Sub chaîne](#ElastiCache-Getting-Started-Tutorials-pub-sub)
+ [Écrire et lire à partir d'un flux](#ElastiCache-Getting-Started-Tutorials-read-write-stream)
### Définir et obtenir des chaînes
Copiez le programme suivant et collez-le dans un fichier nommé *SetAndGetStrings.py.*
```
import time
import logging
logging.basicConfig(level=logging.INFO,format='%(asctime)s: %(message)s')
keyName='mykey'
currTime=time.ctime(time.time())
# Set the key 'mykey' with the current date and time as value.
# The Key will expire and removed from cache in 60 seconds.
redis.set(keyName, currTime, ex=60)
# Sleep just for better illustration of TTL (expiration) value
time.sleep(5)
# Retrieve the key value and current TTL
keyValue=redis.get(keyName)
keyTTL=redis.ttl(keyName)
logging.info("Key {} was set at {} and has {} seconds until expired".format(keyName, keyValue, keyTTL))
```
Pour exécuter le programme, saisissez la commande suivante.
`python SetAndGetStrings.py`
### Définir et obtenir un hachage avec plusieurs éléments
Copiez le programme suivant et collez-le dans un fichier nommé *SetAndGetHash.py.*
```
import logging
import time
logging.basicConfig(level=logging.INFO,format='%(asctime)s: %(message)s')
keyName='mykey'
keyValues={'datetime': time.ctime(time.time()), 'epochtime': time.time()}
# Set the hash 'mykey' with the current date and time in human readable format (datetime field) and epoch number (epochtime field).
redis.hset(keyName, mapping=keyValues)
# Set the key to expire and removed from cache in 60 seconds.
redis.expire(keyName, 60)
# Sleep just for better illustration of TTL (expiration) value
time.sleep(5)
# Retrieves all the fields and current TTL
keyValues=redis.hgetall(keyName)
keyTTL=redis.ttl(keyName)
logging.info("Key {} was set at {} and has {} seconds until expired".format(keyName, keyValues, keyTTL))
```
Pour exécuter le programme, saisissez la commande suivante.
`python SetAndGetHash.py`
### Publier (écrire) et s'abonner (lire) depuis une Pub/Sub chaîne
Copiez le programme suivant et collez-le dans un fichier nommé *PubAndSub.py.*
```
import logging
import time
def handlerFunction(message):
"""Prints message got from PubSub channel to the log output
Return None
:param message: message to log
"""
logging.info(message)
logging.basicConfig(level=logging.INFO)
redis = Redis(host="redis202104053.tihewd.ng.0001.use1.cache.amazonaws.com", port=6379, decode_responses=True)
# Creates the subscriber connection on "mychannel"
subscriber = redis.pubsub()
subscriber.subscribe(**{'mychannel': handlerFunction})
# Creates a new thread to watch for messages while the main process continues with its routines
thread = subscriber.run_in_thread(sleep_time=0.01)
# Creates publisher connection on "mychannel"
redis.publish('mychannel', 'My message')
# Publishes several messages. Subscriber thread will read and print on log.
while True:
redis.publish('mychannel',time.ctime(time.time()))
time.sleep(1)
```
Pour exécuter le programme, saisissez la commande suivante.
`python PubAndSub.py`
### Écrire et lire à partir d'un flux
Copiez le programme suivant et collez-le dans un fichier nommé *ReadWriteStream.py.*
```
from redis import Redis
import redis.exceptions as exceptions
import logging
import time
import threading
logging.basicConfig(level=logging.INFO)
def writeMessage(streamName):
"""Starts a loop writting the current time and thread name to 'streamName'
:param streamName: Stream (key) name to write messages.
"""
fieldsDict={'writerId':threading.currentThread().getName(),'myvalue':None}
while True:
fieldsDict['myvalue'] = time.ctime(time.time())
redis.xadd(streamName,fieldsDict)
time.sleep(1)
def readMessage(groupName=None,streamName=None):
"""Starts a loop reading from 'streamName'
Multiple threads will read from the same stream consumer group. Consumer group is used to coordinate data distribution.
Once a thread acknowleges the message, it won't be provided again. If message wasn't acknowledged, it can be served to another thread.
:param groupName: stream group were multiple threads will read.
:param streamName: Stream (key) name where messages will be read.
"""
readerID=threading.currentThread().getName()
while True:
try:
# Check if the stream has any message
if redis.xlen(streamName)>0:
# Check if if the messages are new (not acknowledged) or not (already processed)
streamData=redis.xreadgroup(groupName,readerID,{streamName:'>'},count=1)
if len(streamData) > 0:
msgId,message = streamData[0][1][0]
logging.info("{}: Got {} from ID {}".format(readerID,message,msgId))
#Do some processing here. If the message has been processed sucessfuly, acknowledge it and (optional) delete the message.
redis.xack(streamName,groupName,msgId)
logging.info("Stream message ID {} read and processed successfuly by {}".format(msgId,readerID))
redis.xdel(streamName,msgId)
else:
pass
except:
raise
time.sleep(0.5)
# Creates the stream 'mystream' and consumer group 'myworkergroup' where multiple threads will write/read.
try:
redis.xgroup_create('mystream','myworkergroup',mkstream=True)
except exceptions.ResponseError as e:
logging.info("Consumer group already exists. Will continue despite the error: {}".format(e))
except:
raise
# Starts 5 writer threads.
for writer_no in range(5):
writerThread = threading.Thread(target=writeMessage, name='writer-'+str(writer_no), args=('mystream',),daemon=True)
writerThread.start()
# Starts 10 reader threads
for reader_no in range(10):
readerThread = threading.Thread(target=readMessage, name='reader-'+str(reader_no), args=('myworkergroup','mystream',),daemon=True)
readerThread.daemon = True
readerThread.start()
# Keep the code running for 30 seconds
time.sleep(30)
```
Pour exécuter le programme, saisissez la commande suivante.
`python ReadWriteStream.py`