Mercurial

comparison asyncio_threads/bucket_questions/single_thread_async.py @ 48:46daba6e3cf4

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Few python scrtips to show how to use asychio.
author MrJuneJune <me@mrjunejune.com>
date Sat, 13 Dec 2025 14:23:02 -0800
parents
children
comparison
equal deleted inserted replaced
47:829623189a57 48:46daba6e3cf4
1 import asyncio
2 from typing import Dict
3
4 class DistributedCache:
5
6 def __init__(self):
7 self._user_bucket_map: Dict[str, TokenBucket] = {}
8
9
10 def get_bucket(self, user_id: str):
11 return self._user_bucket_map[user_id]
12
13 def set_bucket(self, user_id: str):
14 self._user_bucket_map[user_id] = TokenBucket()
15
16
17 INITIAL_VALUES = 10
18
19 class TokenBucket:
20
21 def __init__(self, initial_values = INITIAL_VALUES):
22 self._tokens = initial_values
23 self._refill_values = initial_values
24 self._lock = asyncio.Lock()
25
26 def get_tokens(self):
27 print(self._tokens)
28 return self._tokens
29
30 async def consume_tokens(self, token: int):
31 async with self._lock:
32 if self.get_tokens() < token:
33 return False
34
35 await asyncio.sleep(1)
36 self._tokens -= token
37 return True
38
39 async def refill_tokens(self):
40 async with self._lock:
41 self._tokens = self._refill_values
42
43
44 cache = DistributedCache()
45
46 user_1 = "JUNE"
47 user_2 = "VICTOR"
48 cache.set_bucket(user_1)
49 cache.set_bucket(user_2)
50
51 async def refill_token_bucket(user_id: str):
52 await cache.get_bucket(user_id).refill_tokens()
53
54 async def use_tokens(user_id: str, tokens: int):
55 return await cache.get_bucket(user_id).consume_tokens(tokens)
56
57 async def run():
58 return_value = await asyncio.gather(
59 use_tokens(user_1, 10),
60 refill_token_bucket(user_1),
61 use_tokens(user_1, 10),
62 use_tokens(user_2, 10),
63 )
64 print(return_value)
65
66 if __name__ == "__main__":
67 asyncio.run(run())