fork download
copy 
  1. import asyncio
  2. from typing import Any, Dict
  3.  
  4.  
  5. class ADict:
  6.  
  7.     def __init__(self):
  8.         self.data: Dict[str, asyncio.Future] = {}
  9.         self.lock = asyncio.Lock()
  10.  
  11.     async def get(self, key: str) -> Any:
  12.         if key in self.data:
  13.             async with self.lock:
  14.                 return await self.data[key]
  15.  
  16.         self.data[key] = asyncio.Future()
  17.         return await self.data[key]
  18.  
  19.     async def set(self, key: str, value: Any) -> None:
  20.         async with self.lock:
  21.             if key in self.data:
  22.                 f = self.data[key]
  23.             else:
  24.                 f = asyncio.Future()
  25.             f.set_result(value)
  26.             self.data[key] = f
  27.  
  28.     def __str__(self):
  29.         return f'ADict({self.data})'
  30.  
  31.  
  32. async def show_amount(a_dict: ADict, key: str):
  33.     amount = await a_dict.get(key)
  34.     print(f'GOT: {amount} for {key}')
  35.  
  36.  
  37. async def test():
  38.     print('TEST START')
  39.     a_dict = ADict()
  40.     asyncio.create_task(show_amount(a_dict, 'Moscow'))
  41.     await asyncio.sleep(1)
  42.     await a_dict.set('Moscow', 200000)
  43.     await a_dict.set('Spb', 150000)
  44.     print(a_dict)
  45.     await asyncio.sleep(1)
  46.     asyncio.create_task(show_amount(a_dict, 'Moscow'))
  47.     asyncio.create_task(show_amount(a_dict, 'Spb'))
  48.     await asyncio.sleep(1)
  49.     print(a_dict)
  50.     print('TEST END')
  51.  
  52.  
  53. asyncio.run(test())
Success #stdin #stdout 0.11s 16084KB
comments (?)
stdin
copy 
Standard input is empty
stdout
copy 
TEST START
ADict({'Moscow': <Future finished result=200000>, 'Spb': <Future finished result=150000>})
GOT: 200000 for Moscow
GOT: 200000 for Moscow
GOT: 150000 for Spb
ADict({'Moscow': <Future finished result=200000>, 'Spb': <Future finished result=150000>})
TEST END
https://ideone.com/KvunMu
language:
Python 3 (python  3.9.5)
created:
2 years ago
visibility:
public

Share or Embed source code

Discover > Sphere Engine API

The brand new service which powers Ideone!

Discover > IDE Widget

Widget for compiling and running the source code in a web browser!