Wrap sync into async
Core technique:
data = await asyncio.to_thread(perform_blocking_get, url) is the universal way to put any blocking function inside an async architecture.await asyncio.sleep(1) is used to wait inside aync function. Note that if time.sleep is used, this function basically blocks the execution of other functions.
In this demo:
- A local HTTP server that simulates IN_PROGRESS and DONE (after 4 cumulative requests).
- A client that implements a
async check_status function that polls an HTTP endpoint for current status, it awaits (yields execution to the async loop for other tasks) when IN_PROGRESS is received, and returns when DONE is received.
Code
Below is the all in one code, that is good to execute directly.
import asyncio
import requests
from typing import Dict, Any
def perform_blocking_get(url: str) -> Dict[str, Any]:
"""
A regular (synchronous) function that performs the blocking GET request.
This is the function we will run in a separate thread.
"""
try:
response = requests.get(url, timeout=10)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"A requests error occurred: {e}")
raise
async def check_status(url: str, timeout: int = 60) -> Dict[str, Any]:
"""
Polls an HTTP endpoint using blocking `requests` in a separate thread.
"""
print(f"Starting to poll endpoint: {url} (using blocking requests)")
start_time = asyncio.get_running_loop().time()
while True:
elapsed_time = asyncio.get_running_loop().time() - start_time
if elapsed_time >= timeout:
raise asyncio.TimeoutError(f"Polling timed out after {timeout} seconds.")
try:
print("Checking status...")
data = await asyncio.to_thread(perform_blocking_get, url)
status = data.get("status")
if status == "DONE":
print("Status is DONE. Returning the result.")
return data
elif status == "IN_PROGRESS":
print("Status is IN_PROGRESS. Waiting for 1 second...")
await asyncio.sleep(1)
else:
raise ValueError(f"Received unexpected status: '{status}'")
except Exception as e:
print(f"An error occurred during the poll cycle: {e}")
raise
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
import json
REQUEST_COUNT = 0
class MockHandler(BaseHTTPRequestHandler):
"""A mock handler that simulates a long-running process."""
def do_GET(self):
global REQUEST_COUNT
REQUEST_COUNT += 1
print(f"[Server] Received request #{REQUEST_COUNT}")
self.send_response(200)
self.send_header("Content-type", "application/json")
self.end_headers()
if REQUEST_COUNT < 4:
response_data = {"status": "IN_PROGRESS", "progress": f"{REQUEST_COUNT * 33}%"}
else:
response_data = {"status": "DONE", "result": "Processing complete!", "final_id": 12345}
self.wfile.write(json.dumps(response_data).encode("utf-8"))
def run_server(server_class=HTTPServer, handler_class=MockHandler, port=8080):
server_address = ('', port)
httpd = server_class(server_address, handler_class)
print(f"Starting blocking mock server on port {port}")
httpd.serve_forever()
async def main():
"""Main function to run the server and the client."""
server_thread = threading.Thread(target=run_server)
server_thread.daemon = True
server_thread.start()
await asyncio.sleep(0.1)
endpoint_url = "http://localhost:8080"
try:
final_result = await check_status(endpoint_url, timeout=10)
print("\n--- Polling Finished ---")
print("Final Result:", final_result)
print("------------------------")
except asyncio.TimeoutError:
print("\nPolling failed due to a timeout.")
except Exception as e:
print(f"\nAn error occurred during polling: {e}")
finally:
print("Main function finished. Server thread will exit.")
if __name__ == "__main__":
asyncio.run(main())
Wrap async into sync
The other direction is relatively simple.
Suppose there is an aync func(), one can add await keywords to make it work in the sync environment.
result = await func()
A more curated version is to use callback on the async task.
a = func() -- without the await keyword, a would be a coroutine, instead of actual function return.task = asyncio.create_task(a) -- task.add_done_callback(my_callback) -- set callback.
Good thing is that this allows one to set multiple concurrent tasks and use one await to yield the execution to the main loop in one go.
import asyncio
def my_callback(task):
"""This function is executed when the task finishes."""
print("--- Callback starts ---")
if task.exception():
print(f"The task failed with an exception: {task.exception()}")
else:
result = task.result()
print(f"The task finished successfully. Result: {result}")
print("--- Callback ends ---")
async def func():
"""An example async function that takes time."""
print("func() started...")
await asyncio.sleep(2)
return "Hello from func"
async def main():
print("Calling func() to get the coroutine object.")
a = func()
print("Scheduling the coroutine as a task and adding a callback.")
task = asyncio.create_task(a)
task.add_done_callback(my_callback)
print("main() can do other work now...")
await asyncio.sleep(3)
print("main() is finished.")
