Use DynamoDB as a Cache Backend

This feature gives you a “Redis” liked, serverless, Zero-ops, auto-scaling, high performance, pay-as-you-go cache layer based on DynamoDB.

NOTE: this solution is based on pynamodb_mate Python library.

A lots of developer love Redis. However, redis is not a serverless solution, and you need to manage the server (or cluster). DynamoDB naturally is a serverless, distributive, Key-Value database that has very high read / write throughput. It is a good choice to use DynamoDB as a cache without technique overhead.

⭐ Benefit:

• There’s no server to manage.

• DynamoDB has a latency around 20ms per request.

• DynamoDB cache backend can be created in 5 seconds.

• DynamoDB has pay-as-you-go pricing model, you only pay for what you use.

• DynamoDB automatically scales up and down to adapt your traffic.

• Unlike other local cache solutions, it is on cloud and has permission control and access control out-of-the-box.

Declare a Cache Backend

Cache is nothing but just a key-value store. It uses binary as the storage protocol to store your data (All data will be serialized into binary first). You have to explicitly tell the cache how you convert your data to binary or how you recover your data from binary.

import dataclasses
import time
import pickle
import pynamodb_mate.api as pm

# assume that this is your data
@dataclasses.dataclass
class PersonProfile:
    name: str
    ssn: str

# pm.patterns.cache.DynamoDBBackend[PersonProfile] tells the cache that your data type is PersonProfile
# so you will have type hint automatically
class MyCache(pm.patterns.cache.DynamoDBBackend[PersonProfile]):
    def serialize(self, value: PersonProfile) -> bytes:
        return pickle.dumps(value)

    def deserialize(self, value: bytes) -> PersonProfile:
        return pickle.loads(value)


cache = MyCache(
    table_name="pynamodb-example-cache",
    region="us-east-1", # aws region
    # billing_mode="PAY_PER_REQUEST", # billing model to use when creating the table
    # write_capacity_units=5, # WCU configuration when creating the table
    # read_capacity_units=5, # RCU configuration when creating the table
    # if True, create the table when initializing the backend,
    # if table already exists, then do nothing. if False, you should create
    # the table manually before using the backend.
    create=True,
)

# delete all item in the test table to ensure a fresh start
cache.Table.delete_all();

THAT’s IT, by default it uses pay as you go pricing model.

Use Cache.set and Cache.get

key = "alice"

# cache not hit
print(cache.get(key))

None

cache.set(key, PersonProfile(name="Alice", ssn="123-45-6789"))

cache.get(key)

PersonProfile(name='Alice', ssn='123-45-6789')

Use Time-to-live (TTL)

key = "bob"

# set the key-value with TTL
cache.set(key, PersonProfile(name="Bob", ssn="123-45-6789"), expire=1)

print("we try to get the value immediately, the cache should be hit:")
print(f"{cache.get(key) = }")

print("after wait for 2 seconds, the cache should be expired:")
time.sleep(2)
print(f"{cache.get(key) = }")

we try to get the value immediately, the cache should be hit:
cache.get(key) = PersonProfile(name='Bob', ssn='123-45-6789')
after wait for 2 seconds, the cache should be expired:
cache.get(key) = None