Coverage for src/main.mys : 0%
Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from argparse import Parser
2from redis import ArrayReply
3from redis import BulkStringReply
4from redis import Client
5from redis import ErrorReply
6from redis import IntegerReply
7from redis import Reply
8from redis import SimpleStringReply
10func _print(value: string, indent: string):
11 print(indent, end="")
12 print(value)
14func _print_reply(reply: Reply, indent: string = ""):
15 match reply:
16 case ErrorReply() as error_reply:
17 _print(f"{error_reply.error}: {error_reply.message}", indent)
18 case SimpleStringReply() as simple_string_reply:
19 _print(simple_string_reply.value, indent)
20 case BulkStringReply() as bulk_string_reply:
21 if bulk_string_reply.value is None:
22 _print("nil", indent)
23 else:
24 _print(str(bulk_string_reply.value), indent)
25 case IntegerReply() as integer_reply:
26 _print(str(integer_reply.value), indent)
27 case ArrayReply() as array_reply:
28 _print("(array)", indent)
30 if array_reply.items is None:
31 _print("nil", indent)
32 else:
33 for item in array_reply.items:
34 _print_reply(item, indent + " ")
36func main(argv: [string]):
37 parser = Parser()
38 parser.add_option("--host",
39 default="127.0.0.1",
40 help="Redis server host (default: 127.0.0.1).")
41 parser.add_option("--port",
42 default="6379",
43 help="Redis server port (default: 6379)")
44 parser.add_positional("command",
45 multiple_occurrences=True,
46 help="Command to execute.")
47 args = parser.parse(argv)
49 client = Client()
50 client.connect(args.value_of("--host"), i64(args.value_of("--port")))
52 command = [arg.to_utf8() for arg in args.values_of("command")]
53 _print_reply(client.call(command))
55 if string(command[0]).upper() in ["SUBSCRIBE", "PSUBSCRIBE"]:
56 while True:
57 print(client.get_message())
59 client.disconnect()