Use the asyncio module for asynchronous requests in Python. Let's create code that hits the API asynchronously using the requests package.
API
I could have used time.sleep ()
, but this time I will assume a pattern to get the API rather than practice.
I used JSON Placeholder
as the mock server.
Write the code that requests the APIs to get the list (six below) synchronously and asynchronously, and returns the result of each result in one list.
https://jsonplaceholder.typicode.com/posts https://jsonplaceholder.typicode.com/comments https://jsonplaceholder.typicode.com/albums https://jsonplaceholder.typicode.com/photos https://jsonplaceholder.typicode.com/todos https://jsonplaceholder.typicode.com/users
First, let's make a request in the synchronous process. If you write it according to the usual Python code, it will be as follows.
import time
import requests
BASE_URL = "https://jsonplaceholder.typicode.com/"
def calc_time(fn):
"""Decorator to measure function execution time"""
def wrapper(*args, **kwargs):
start = time.time()
fn(*args, **kwargs)
end = time.time()
print(f"[{fn.__name__}] elapsed time: {end - start}")
return
return wrapper
def get_sync(path: str) -> dict:
print(f"/{path} request")
res = requests.get(BASE_URL + path)
print(f"/{path} request done")
return res.json()
@calc_time
def main_sync():
data_ls = []
paths = [
"posts",
"comments",
"albums",
"photos",
"todos",
"users",
]
for path in paths:
data_ls.append(get_sync(path))
return data_ls
if __name__ == "__main__":
main_sync()
When run, you will get output similar to the following:
Request 1-> Request 1 completed Request 2-> Request 2 completed Request 3-> Request 3 Completed ...
You can see that it is being executed.
/posts request
/posts request done
/comments request
/comments request done
/albums request
/albums request done
/photos request
/photos request done
/todos request
/todos request done
/users request
/users request done
[main_sync] elapsed time: 1.157785415649414
Then make an asynchronous request.
Asynchronous tasks run inside an event loop. To get the event loop, use ʻasyncio.get_event_loop () `.
loop.run_until_complete
, as the name implies, is a method that executes an event loop until each task has been executed.
The return value of this method is a list containing the return value of each asynchronous execution task.
The order of execution is not guaranteed, but the return values are returned in the order passed to the argument, so it can be used even when the order is important.
The get_async
declared with ʻasync def is called a coroutine function. The ʻawait
expression inside a coroutine function pauses the execution of the coroutine function until the return value is returned.
import asyncio
#Coroutine function
async def get_async(path: str) -> dict:
print(f"/{path} async request")
url = BASE_URL + path
loop = asyncio.get_event_loop()
#Run in event loop
res = await loop.run_in_executor(None, requests.get, url)
print(f"/{path} async request done")
return res.json()
@calc_time
def main_async():
#Get event loop
loop = asyncio.get_event_loop()
#Asynchronous execution task in one Future object
tasks = asyncio.gather(
get_async("posts"),
get_async("comments"),
get_async("albums"),
get_async("photos"),
get_async("todos"),
get_async("users"),
)
#Asynchronous execution, until each is finished
results = loop.run_until_complete(tasks)
return results
if __name__ == "__main__":
main_async()
The output looks like this:
Request 1 Request 2 Request 3 ... Request 1 completed Request 2 completed Request 3 completed ...
You can see that it is processed. You can also see that the execution time has been significantly reduced.
/posts async request
/comments async request
/albums async request
/photos async request
/todos async request
/users async request
/users async request done
/todos async request done
/posts async request done
/albums async request done
/comments async request done
/photos async request done
[main_async] elapsed time: 0.17921733856201172
Recommended Posts