Memcached¶
Since testcontainers-python v4.7.0
Introduction¶
The Testcontainers module for Memcached.
Adding this module to your project dependencies¶
Please run the following command to add the Memcached module to your python dependencies:
pip install testcontainers[memcached] pymemcache
Usage example¶
import json
import pickle
import memcache
from testcontainers.memcached import MemcachedContainer
def basic_example():
with MemcachedContainer() as memcached:
# Get connection parameters
host = memcached.get_container_host_ip()
port = memcached.get_exposed_port(memcached.port)
# Create Memcached client
client = memcache.Client([f"{host}:{port}"])
print("Connected to Memcached")
# Store simple values
client.set("string_key", "Hello from Memcached")
client.set("int_key", 42)
client.set("float_key", 3.14)
print("Stored simple values")
# Store complex data
complex_data = {"name": "test", "values": [1, 2, 3], "nested": {"key": "value"}}
client.set("complex_key", json.dumps(complex_data))
print("Stored complex data")
# Store with expiration
client.set("expiring_key", "This will expire", time=5)
print("Stored value with expiration")
# Store with pickle
class TestObject:
def __init__(self, name, value):
self.name = name
self.value = value
test_obj = TestObject("test", 123)
client.set("object_key", pickle.dumps(test_obj))
print("Stored pickled object")
# Retrieve values
print("\nRetrieved values:")
print(f"string_key: {client.get('string_key')}")
print(f"int_key: {client.get('int_key')}")
print(f"float_key: {client.get('float_key')}")
# Retrieve complex data
complex_value = json.loads(client.get("complex_key"))
print("\nComplex data:")
print(json.dumps(complex_value, indent=2))
# Retrieve pickled object
obj_data = pickle.loads(client.get("object_key"))
print("\nPickled object:")
print(f"name: {obj_data.name}")
print(f"value: {obj_data.value}")
# Check expiration
print("\nChecking expiring key:")
print(f"expiring_key: {client.get('expiring_key')}")
print("Waiting for key to expire...")
import time
time.sleep(6)
print(f"expiring_key after expiration: {client.get('expiring_key')}")
# Store multiple values
multi_data = {"key1": "value1", "key2": "value2", "key3": "value3"}
client.set_multi(multi_data)
print("\nStored multiple values")
# Retrieve multiple values
multi_keys = ["key1", "key2", "key3"]
multi_values = client.get_multi(multi_keys)
print("\nMultiple values:")
print(json.dumps(multi_values, indent=2))
# Increment and decrement
client.set("counter", 0)
client.incr("counter")
client.incr("counter", 2)
print("\nCounter after increment:")
print(f"counter: {client.get('counter')}")
client.decr("counter")
print("Counter after decrement:")
print(f"counter: {client.get('counter')}")
# Store with flags
client.set("flagged_key", "value", flags=1)
print("\nStored value with flags")
# Get stats
stats = client.get_stats()
print("\nMemcached stats:")
for server, server_stats in stats:
print(f"\nServer: {server}")
print(json.dumps(dict(server_stats), indent=2))
# Delete values
client.delete("string_key")
client.delete_multi(["key1", "key2", "key3"])
print("\nDeleted values")
# Check deleted values
print("\nChecking deleted values:")
print(f"string_key: {client.get('string_key')}")
print(f"key1: {client.get('key1')}")
# Store with CAS
client.set("cas_key", "initial")
cas_value = client.gets("cas_key")
print("\nCAS value:")
print(f"value: {cas_value}")
# Update with CAS
success = client.cas("cas_key", "updated", cas_value[1])
print(f"CAS update success: {success}")
print(f"Updated value: {client.get('cas_key')}")
# Try to update with invalid CAS
success = client.cas("cas_key", "failed", 0)
print(f"Invalid CAS update success: {success}")
print(f"Value after failed update: {client.get('cas_key')}")
# Clean up
client.flush_all()
print("\nFlushed all values")
if __name__ == "__main__":
basic_example()