This concise, straight-to-the-point article is about using async/await
with class methods in Python. We’ll have a look at the fundamentals and then walk through some practical code examples.
What is the Point?
Adding an asynchronous method to a class is almost the same as creating a normal asynchronous function. Below are the steps you need to follow:
- Define your class method with theÂ
async
 keyword beforeÂdef
. This means that the method is a coroutine, which is a special kind of function that can be paused and resumed by theÂawait
 keyword. - Inside your class method, use theÂ
await
 keyword before any expression that returns another coroutine or an awaitable object. This means that the execution of your method will be suspended until the awaited expression is done, and then resume with the result of that expression. - To call your class method, you need to use theÂ
await
 keyword before it as well. This means that you can only call it from another coroutine or anasync
context manager. You also need to use an event loop, which is an object that manages the execution of coroutines and other asynchronous tasks.
Note: You cannot use await
in the __init__
method of a class, because it is not a coroutine, and it must return None
.
Examples
Data Processing
In this example, we will define a class that mimics data processing tasks:
# SlingAcademy.com
# This code uses Python 3.11.4
import asyncio
# Define a class that models a data processor
class DataProcessor:
def __init__(self, data):
self.data = data
async def process_data(self):
print("Start processing data...")
await asyncio.sleep(2) # Simulate an asynchronous operation
processed_data = [item * 2 for item in self.data]
print("Data processing complete!")
return processed_data
# Define a main() coroutine that uses the DataProcessor class
async def main():
# Create a list of data
data = [1, 2, 3, 4, 5]
# Create an instance of the DataProcessor class
processor = DataProcessor(data)
# Process the data
processed_data = await processor.process_data()
print("Processed data:", processed_data)
if __name__ == "__main__":
asyncio.run(main())
In this example, we define a DataProcessor
class with an asynchronous method process_data()
. Inside the method, we use await
to pause the execution while simulating an asynchronous operation with asyncio.sleep(2)
. Once the processing is complete, we return the processed data.
In the main()
function, we create an instance of the DataProcessor
class and call the process_data()
method using await
to wait for the result. Finally, we print the processed data.
You’ll receive the following output:
Start processing data...
Data processing complete!
Processed data: [2, 4, 6, 8, 10]
Fetching Data from a REST API
In this example, we’ll define a class that has an async method that makes a web request using aiohttp
(a third-party library for asynchronous HTTP requests). You can install aiohttp
by running this command:
pip install aiohttp
Example code:
# SlingAcademy.com
# This code uses Python 3.11.4
import asyncio
import aiohttp
class WebFetcher:
# Define an async method with the async keyword
async def fetch(self, url):
# Use a context manager (async with) to create a session
async with aiohttp.ClientSession() as session:
# Use await to call another coroutine (session.get)
async with session.get(url) as response:
# Use await to read the response content as text
data = await response.json()
# Print the response status and text
print(f"Status: {response.status}")
print(f"data: {data}")
# Create an instance of the class
fetcher = WebFetcher()
# Get the default event loop
loop = asyncio.get_event_loop()
# Run the async method using the event loop
loop.run_until_complete(fetcher.fetch("https://api.slingacademy.com"))
Output:
Status: 200
data: {'success': True, 'message': 'Welcome to Sling Academy Public API'}
In case you want to practice more about making HTTP requests with aiohttp
, just try our public REST API endpoints listed below:
- Sample Blog Posts – Public REST API for Practice
- Sample Users – Free Fake API for Practicing & Prototyping
- Sample Users – Free Fake API for Practicing & Prototyping
- Sample Blog Posts – Public REST API for Practice
Common Pitfalls & Best Practices
Using async/await
in Python classes can be a powerful way to write concurrent and responsive code, but it also comes with some challenges and pitfalls that you should be aware of. Here are some of the common ones and how to avoid them:
- Don’t mix synchronous and asynchronous code. If you have a class that has both synchronous and asynchronous methods, you might run into problems when calling them from different contexts. For example, if you call a synchronous method from an asynchronous one, you will block the event loop and prevent other coroutines from running. Conversely, if you call an asynchronous method from a synchronous one, you will get a coroutine object instead of the actual result, and you will need to use an event loop to run it. To avoid these issues, you should either make all your methods synchronous or asynchronous, or use different classes for different contexts.
- Don’t forget to await your coroutines. As we saw in the previous example, if you don’t use the await keyword when calling an asynchronous method, you will get a coroutine object that has not been executed yet. This can lead to unexpected behavior and errors, especially if you try to use the coroutine object as a regular value. To avoid this, you should always use await when calling an asynchronous method, or store the coroutine object in a variable and await it later.
- Don’t use blocking calls in your coroutines. If you use a blocking call, such as
time.sleep()
orinput()
, in your coroutine, you will also block the event loop and prevent other coroutines from running. This defeats the purpose of using async/await, which is to make your code more concurrent and responsive. To avoid this, you should use non-blocking alternatives, such asasyncio.sleep()
orasyncio.get_event_loop().run_in_executor()
, which will yield control back to the event loop while waiting for the result. - Don’t forget to handle exceptions. Just like in regular code, exceptions can occur in your asynchronous code, and you need to handle them properly. If an exception is raised in your coroutine and you don’t catch it, it will propagate to the caller and eventually to the event loop, which might terminate your program or log an error message. To avoid this, you should use
try/except
blocks in your coroutines, or useasyncio.gather()
orasyncio.wait()
with thereturn_exceptions
parameter set toTrue
when running multiple coroutines concurrently.
That’s it. This tutorial ends here. Goodbye and see you again in other articles on Sling Academy!