Async Advanced

advanced.py
 1import asyncio
 2
 3from zcached.asyncio import AsyncZCached
 4from zcached import Result
 5from typing import TypedDict, List
 6
 7
 8class Item(TypedDict):
 9    name: str
10    description: str
11    price: float
12    quantity: int
13
14
15class ShopManager(AsyncZCached):
16
17    async def set_items(self, items: List[Item]) -> str:
18        if not await self.is_alive():
19            raise RuntimeError("Something went wrong :(")
20
21        result: Result[str] = await self.set(key="items", value=items)
22        if not result:
23            raise RuntimeError(result.error)
24
25        return result.value
26
27    async def get_items(self) -> List[Item]:
28        if not self.is_alive():
29            raise RuntimeError("Connection closed.")
30
31        result: Result[List[Item]] = await self.get(key="items")
32        if not result:
33            raise RuntimeError(result.error)
34
35        return result.value
36
37    async def get_total_quantity(self) -> int:
38        return sum([item["quantity"] for item in await self.get_items()])
39
40    async def get_total_cost(self) -> float:
41        return sum([item["price"] for item in await self.get_items()])
42
43
44async def main():
45    manager = ShopManager(host="127.0.0.1", port=7556)
46    await manager.run()
47    await manager.set_items(
48        [
49            Item(name="foo", description="test", price=50.99, quantity=1),
50            Item(name="bar", description="test123", price=89.99, quantity=5),
51        ]
52    )
53    print(await manager.get_items())
54    print(await manager.get_total_quantity())
55    print(await manager.get_total_cost())
56    await manager.flush()
57
58
59if __name__ == "__main__":
60    asyncio.run(main())