Skip to content

Instantly share code, notes, and snippets.

@messa
Last active June 19, 2020 12:23
Show Gist options
  • Save messa/1fa79c8456eb4abecc2fdf29f625e8cb to your computer and use it in GitHub Desktop.
Save messa/1fa79c8456eb4abecc2fdf29f625e8cb to your computer and use it in GitHub Desktop.
#!/usr/bin/env python3
import asyncio
import logging
from pprint import pprint
from reprlib import repr as smart_repr
import requests
logger = logging.getLogger(__name__)
async def main():
logging.basicConfig(level=logging.DEBUG)
urls = '''
https://reqres.in/api/users/1
https://reqres.in/api/users/2
https://reqres.in/api/users/3
https://reqres.in/api/users/4
https://reqres.in/api/users/5
https://reqres.in/api/users/6
https://reqres.in/api/users/7
https://reqres.in/api/users/8
https://reqres.in/api/users/9
https://reqres.in/api/users/10
'''.split()
users = {}
tasks = [asyncio.create_task(retrieve_users(urls, users)) for i in range(4)]
await asyncio.wait(tasks)
pprint(users)
async def retrieve_users(urls, users):
loop = asyncio.get_running_loop()
while urls:
url = urls.pop()
try:
# funkce retrieve_user není async, takže ji pustíme přes threadpool executor
user = await loop.run_in_executor(None, retrieve_user, url)
except Exception as e:
logger.error('Failed to download user from %s: %r', url, e)
else:
users[user['id']] = user
def retrieve_user(url):
logger.debug('Downloading %s', url)
response = requests.get(url)
response.raise_for_status()
data = response.json()
logger.debug('Downloaded %s -> %s', url, smart_repr(data))
return data['data']
if __name__ == '__main__':
asyncio.run(main())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment