testcontainers-python/modules/valkey/example_basic.py at 2108a2081442a0d179ba78a8ce826ba2fe54b97b · daric93/testcontainers-python · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
"""
Valkey container usage examples with valkey-glide sync client.

Requires: pip install valkey-glide-sync
"""

from glide_sync import GlideClient, GlideClientConfiguration, NodeAddress, ServerCredentials

from testcontainers.valkey import ValkeyContainer


def basic_example():
    with ValkeyContainer() as valkey_container:
        host = valkey_container.get_host()
        port = valkey_container.get_exposed_port()
        connection_url = valkey_container.get_connection_url()

        print(f"Valkey connection URL: {connection_url}")
        print(f"Host: {host}, Port: {port}")

        config = GlideClientConfiguration([NodeAddress(host, port)])
        client = GlideClient.create(config)

        pong = client.ping()
        print(f"PING response: {pong}")

        client.set("key", "value")
        print("SET response: OK")

        value = client.get("key")
        print(f"GET response: {value}")

        client.close()


def password_example():
    with ValkeyContainer().with_password("mypassword") as valkey_container:
        host = valkey_container.get_host()
        port = valkey_container.get_exposed_port()
        connection_url = valkey_container.get_connection_url()

        print(f"\nValkey with password connection URL: {connection_url}")

        config = GlideClientConfiguration(
            [NodeAddress(host, port)],
            credentials=ServerCredentials(password="mypassword"),
        )
        client = GlideClient.create(config)

        pong = client.ping()
        print(f"PING response: {pong}")

        client.close()


def version_example():
    with ValkeyContainer().with_image_tag("8.0") as valkey_container:
        print(f"\nUsing image: {valkey_container.image}")
        connection_url = valkey_container.get_connection_url()
        print(f"Connection URL: {connection_url}")


def bundle_example():
    with ValkeyContainer().with_bundle() as valkey_container:
        print(f"\nUsing bundle image: {valkey_container.image}")
        host = valkey_container.get_host()
        port = valkey_container.get_exposed_port()

        config = GlideClientConfiguration([NodeAddress(host, port)])
        client = GlideClient.create(config)

        pong = client.ping()
        print(f"PING response: {pong}")

        client.close()


if __name__ == "__main__":
    basic_example()
    password_example()
    version_example()
    bundle_example()