Python: Using async/await with class methods (3 examples)

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:

1. 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.
2. 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.
3. 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 an async 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() or input(), 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 as asyncio.sleep() or asyncio.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 use asyncio.gather() or asyncio.wait() with the return_exceptions parameter set to True when running multiple coroutines concurrently.

That’s it. This tutorial ends here. Goodbye and see you again in other articles on Sling Academy!