Amazon Keyspaces (for Apache Cassandra) のキャッシュとしての Amazon ElastiCache の利用

from cassandra.cluster import *
 from ssl import SSLContext, PROTOCOL_TLSv1_2 , CERT_REQUIRED 
 from cassandra.auth import PlainTextAuthProvider 
 from cassandra_sigv4.auth import SigV4AuthProvider 
 from cassandra.query import SimpleStatement 
 import time 
 import boto3 

 ssl_context = SSLContext(PROTOCOL_TLSv1_2)
 ssl_context.load_verify_locations('/home/ec2-user/sf-class2-root.crt')
 ssl_context.verify_mode = CERT_REQUIRED 
 boto_session = boto3.Session()
 auth_provider = SigV4AuthProvider(boto_session)
 cluster = Cluster(['cassandra.us-east-1.amazonaws.com'],
                  ssl_context=ssl_context,
                  auth_provider=auth_provider,
                  port=9142)
 session = cluster.connect()

from rediscluster import RedisCluster 
 import logging 

 logging.basicConfig(level=logging.ERROR)
 redis = RedisCluster(startup_nodes=[{"host": "keyspaces-cache.eeeccc.clustercfg.use1.cache.amazonaws.com",
                                     "port": "6379"}],
                     decode_responses=True,
                     skip_full_coverage_check=True)
 if redis.ping():
    logging.info("Connected to Redis")

#Global variables 
 keyspace_name="catalog"
 table_name="book_awards"


#Write a row 
 def write_award(book_award):
    write_award_to_keyspaces(book_award)
    write_award_to_cache(book_award)

#write row to the Keyspaces table 
 def write_award_to_keyspaces(book_award):
    stmt=SimpleStatement(f"INSERT INTO {keyspace_name}.{table_name} (award, year, category, rank, author, book_title, publisher) VALUES (%s, %s, %s, %s, %s, %s, %s)",
                         consistency_level=ConsistencyLevel.LOCAL_QUORUM)
    session.execute(stmt,(book_award["award"],
                                book_award["year"],
                                book_award["category"],
                                book_award["rank"],
                                book_award["author"],
                                book_award["book_title"],
                                book_award["publisher"]))

#write row to the cache 
 def write_award_to_cache(book_award):
    #construct Redis key name 
    key_name=book_award["award"] + str(book_award["year"])+ book_award["category"] + str(book_award["rank"])

    #write to cache using Redis set, ex=300 sets TTL for this row to 300 seconds 
    redis.set(key_name, str(book_award), ex=300)


#Delete a row 
 def delete_award(award, year, category, rank):
    delete_award_from_keyspaces(award, year, category, rank)
    delete_award_from_cache(award, year, category, rank)

#delete row from Keyspaces table 
 def delete_award_from_keyspaces(award, year, category, rank):
    stmt = SimpleStatement(f"DELETE FROM {keyspace_name}.{table_name} WHERE award=%s AND year=%s AND category=%s AND rank=%s ;",
                           consistency_level=ConsistencyLevel.LOCAL_QUORUM)
    session.execute(stmt, (award, int(year), category, int(rank)))

#delete row from cache 
 def delete_award_from_cache(award, year, category, rank):
    #construct Redis key name    
    key_name=award + str(year)+ category + str(rank)

    #delete the row from cache if it exists 
    if redis.get(key_name) is not None:
        book_award=redis.delete(key_name)

#Global variables 
 keyspace_name="catalog"
 table_name="book_awards"


#Read a row 
 def get_award(award, year, category, rank):
    #construct Redis key name from parameters 
    key_name=award + str(year)+ category + str(rank)
    start=time.time()
    book_award=redis.get(key_name)
    end=time.time()
    elapsed=(end - start)*1000 

    #if row not in cache, fetch it from Keyspaces table 
    if not book_award:
        print("Fetched from Cache: ", book_award)
        stmt = SimpleStatement(f"SELECT * FROM {keyspace_name}.{table_name} WHERE award=%s AND year=%s AND category=%s AND rank=%s ;")
        start=time.time()
        res=session.execute(stmt, (award, int(year), category, int(rank)))
        end=time.time()
        elapsed=(end - start)*1000 
        if not res.current_rows:
            print("Fetched from Database: None")
            return None 
        else:
            #lazy-load into cache 
            book_award=redis.set(key_name, str(res.current_rows), ex=300)
            print("Fetched from Database in: ", elapsed, "ms")
            return res.current_rows 
    else:
        print("Fetched from Cache in: ", elapsed, "ms")
        return book_award 

#Global variables 
 keyspace_name="catalog"
 table_name="book_awards"


#Read one or more rows based on parameters 
 def get_awards(parameters):
    #construct key name from parameters 
    key_name=""
    for p in parameters:
        key_name=key_name + str(p)
    
    start=time.time()
    book_awards=redis.lrange(key_name, 0, -1)
    end=time.time()
    elapsed=(end - start)*1000 

    #if result set not in cache, fetch it from Keyspaces table 
    if not book_awards:
        print("Fetched from Cache: ", book_awards)
        stmt = SimpleStatement(f"SELECT * FROM {keyspace_name}.{table_name} WHERE award=%s AND year=%s AND category=%s AND rank<=%s ;")
        start = time.time()
        res=session.execute(stmt, parameters)
        end=time.time()
        elapsed=(end - start)*1000 
        if not res.current_rows:
            print("Fetched from Database: None")
            return None 
        else:
            #lazy-load into cache 
            redis.rpush(key_name, str(res.current_rows))
            redis.expire(key_name, 300)
            print("Fetched from Database in: ", elapsed, "ms")
            return res.current_rows 
    else:
        print("Fetched from Cache: ", elapsed, "ms")
        return book_awards 

def test_case_1():
    book_award={"award": "Golden Read",
                "year": 2021,
                "category": "sci-fi",
                "rank": 2,
                "author": "John Doe",
                "book_title": "Tomorrow is here",
                "publisher": "Ostrich books"}

    #insert an award to the DB and cache
    write_award(book_award)
    print("Test Case 1:")
    print("New book award inserted.")

    #cache hit - get award from cache
    print("\n")
    print("Verify cache hit:")
    res=get_award(book_award["award"],
              book_award["year"],
              book_award["category"],
              book_award["rank"])
    print(res)

    #let the cache entry expire
    print("\n")
    print("Waiting for cached entry to expire, sleeping for 300 seconds...")
    time.sleep(300)

    #cache miss - get award from DB and lazy load to cache
    print("\n")
    print("Entry expired in cache, award expected to be fetched from DB:")
    res=get_award(book_award["award"],
              book_award["year"],
              book_award["category"],
              book_award["rank"])
    print(res)

    #cache hit - get award from cache
    print("\n")
    print("Verify that award is lazy loaded into cache:")
    res=get_award(book_award["award"],
              book_award["year"],
              book_award["category"],
              book_award["rank"])
    print(res)

    #delete the award from cache and DB
    print("\n")
    print("Deleting book award.")
    delete_award(book_award["award"],
                 book_award["year"],
                 book_award["category"],
                 book_award["rank"])

    #confirm the award was deleted from cache and DB
    print("\n")
    print("Verify that the award was deleted from cache and DB:")
    res=get_award(book_award["award"],
              book_award["year"],
              book_award["category"],
              book_award["rank"])
    if res:
        print(res)

Test Case 1:
 New book award inserted.


 Verify cache hit:
 Fetched from Cache in:  0.3809928894042969 ms 
{'award': 'Golden Read', 'year': 2021, 'category': 'sci-fi', 'rank': 2, 'author': 'John Doe', 'book_title': 'Tomorrow is here', 'publisher': 'Ostrich books'}


 Waiting for cached entry to expire, sleeping for 300 seconds...


 Entry expired in cache, award expected to be fetched from DB:
 Fetched from Cache:  None 
 Fetched from Database in:  14.202594757080078 ms 
 [Row(year=2021, award='Golden Read', category='sci-fi', rank=2, author='John Doe', book_title='Tomorrow is here', publisher='Ostrich books')] 


 Verify that award is lazy loaded into cache:
 Fetched from Cache in:  0.4191398620605469 ms 
 [Row(year=2021, award='Golden Read', category='sci-fi', rank=2, author='John Doe', book_title='Tomorrow is here', publisher='Ostrich books')] 


 Deleting book award.


 Verify that the award was deleted from cache and DB:
 Fetched from Cache:  None 
 Fetched from Database: None 

def test_case_2():
    print("Test Case 2:")
    print("Get top 3 Must Read book awards for year 2021 in the Sci-Fi category")
    print("\n")
    res=get_awards(["Must Read", 2021, "Sci-Fi", 3])
    print(res)

    #cache-hit - get awards from cache 
    print("\n")
    print("Verify cache hit on subsequent query with same parameters:")
    res=get_awards(["Must Read", 2021, "Sci-Fi", 3])
    print(res)

    #let the cache entry expire 
    print("\n")
    print("Waiting for cached entry to expire, sleeping for 300 seconds...")
    time.sleep(300)

    #cache miss - get award from DB and lazy load to cache 
    print("\n")
    print("Entry expired in cache, awards expected to be fetched from DB.")
    res=get_awards(["Must Read", 2021, "Sci-Fi", 3])
    print(res)

    #cache hit - get award from cache 
    print("\n")
    print("Verify that awards are lazy loaded into cache:")
    res=get_awards(["Must Read", 2021, "Sci-Fi", 3])
    print(res)

Test Case 2:
Get top 3 Must Read book awards for year 2021 in the Sci-Fi category


Fetched from Cache:  []
Fetched from Database in:  21.03400230407715 ms
[Row(year=2021, award='Must Read', category='Sci-Fi', rank=1, author='Polly Gon', book_title='The mystery of the 7th dimension', publisher='PublishGo'), Row(year=2021, award='Must Read', category='Sci-Fi', rank=2, author='Kai K', book_title='Time travellers guide', publisher='Publish123'), Row(year=2021, award='Must Read', category='Sci-Fi', rank=3, author='Mick Key', book_title='Key to the teleporter', publisher='Penguins')]


Verify cache hit on subsequent query with same parameters:
Fetched from Cache:  0.36835670471191406 ms
["[Row(year=2021, award='Must Read', category='Sci-Fi', rank=1, author='Polly Gon', book_title='The mystery of the 7th dimension', publisher='PublishGo'), Row(year=2021, award='Must Read', category='Sci-Fi', rank=2, author='Kai K', book_title='Time travellers guide', publisher='Publish123'), Row(year=2021, award='Must Read', category='Sci-Fi', rank=3, author='Mick Key', book_title='Key to the teleporter', publisher='Penguins')]"]


Waiting for cached entry to expire, sleeping for 300 seconds...


Entry expired in cache, awards expected to be fetched from DB.
Fetched from Cache:  []
Fetched from Database in:  32.64594078063965 ms
[Row(year=2021, award='Must Read', category='Sci-Fi', rank=1, author='Polly Gon', book_title='The mystery of the 7th dimension', publisher='PublishGo'), Row(year=2021, award='Must Read', category='Sci-Fi', rank=2, author='Kai K', book_title='Time travellers guide', publisher='Publish123'), Row(year=2021, award='Must Read', category='Sci-Fi', rank=3, author='Mick Key', book_title='Key to the teleporter', publisher='Penguins')]


Verify that awards are lazy loaded into cache:
Fetched from Cache:  0.3757476806640625 ms
["[Row(year=2021, award='Must Read', category='Sci-Fi', rank=1, author='Polly Gon', book_title='The mystery of the 7th dimension', publisher='PublishGo'), Row(year=2021, award='Must Read', category='Sci-Fi', rank=2, author='Kai K', book_title='Time travellers guide', publisher='Publish123'), Row(year=2021, award='Must Read', category='Sci-Fi', rank=3, author='Mick Key', book_title='Key to the teleporter', publisher='Penguins')]"]