Asyncio Coroutine Object Methods in Python

We can define coroutine methods on custom Python objects.

This allows methods on custom Python objects to use async/await syntax, such as awaiting other coroutines and tasks and allows the custom coroutine methods themselves to be awaited within our asyncio programs.

In this tutorial, you will discover how to define object methods as coroutines.

Let's get started.

What is a Coroutine

A coroutine represents a special type of function that can pause its execution at specific points without blocking other tasks.

It allows for concurrent and non-blocking operations, enabling asynchronous programming.

Coroutines are declared using the "async def" expression in Python, distinguishing them from regular functions.

For example:

# define a coroutine
async def coro():
	# ...

They can be paused or suspended using the await expression within their body, allowing other coroutines or tasks to run while waiting for potentially blocking operations like I/O or delays.

Coroutines are a fundamental building block in asynchronous programming with Python, enabling efficient handling of concurrent tasks without resorting to thread-based parallelism.

They facilitate cooperative multitasking and make it easier to write asynchronous code that remains responsive and scalable.

You can learn more about coroutines in the tutorial:

• What is a Coroutine in Python

How to Define Object Methods as Coroutines

We can define object methods as coroutines.

Recall, a method is a function that is defined in a class and has access to the object instance via an argument called self.

method: A function which is defined inside a class body. If called as an attribute of an instance of that class, the method will get the instance object as its first argument (which is usually called self).

-- Python Glossary

For example:

# custom class
class MyObject(object):
	def custom_method(self):
		pass

The method can then be called on the object.

For example:

...
# create the object
obj = MyObject()
# call the method
obj.custom_method()

We can define methods that are coroutines.

This means that they return coroutine objects that can be awaited.

For example, we can define a method as a coroutine using the "async def" expression.

As a coroutine, it means the method can then await other coroutines and tasks.

# custom class
class MyObject(object):
	async def custom_method(self):
		# await another coroutine
		await asyncio.sleep(1)

We can then create the object and await the coroutine within our asyncio program.

For example:

...
# create the object
obj = MyObject()
# await the coroutine
await obj.custom_method()

A coroutine method must be awaited within an asyncio program.

If it is called in a regular program, it will result in a RuntimeWarinng as the created coroutine object will not be executed.

You can learn more about this in the tutorial:

• RuntimeWarning: Coroutine Was Never Awaited

Now that we know how to define object methods as coroutines, let's look at some worked examples.

Example of Object Method As Coroutine

We can explore an example of defining an object method as a coroutine.

In this example, we will define a custom class that has a constructor, a regular method, and a coroutine method. We will then create and use this object within our asyncio program.

Firstly, we can define our custom class.

The constructor of the custom class will take a data argument and store it as an attribute or member variable.

The class will provide a regular method that reports the value of the data attribute with a print statement. It also provides a coroutine method that awaits the sleep coroutine for a moment and then reports the value of the data attribute with a print statement.

The CustomClass below implements this.

# define a custom class
class CustomClass(object):
    # constructor
    def __init__(self, data):
        # store the data
        self.data = data

    # regular method
    def report_data(self):
        # report the value of data
        print(f'Method: {self.data}')

    # method as coroutine
    async def sleep_and_report_data(self):
        # suspend a moment
        await asyncio.sleep(2)
        # report the data
        print(f'Coroutine: {self.data}')

Next, we can define the main coroutine for our asyncio program.

This will create our CustomClass instance and provide it with a string value that will be stored in the data attribute. We will then call the regular method, then await the coroutine method.

The main() coroutine below implements this.

# main coroutine
async def main():
    # create the custom object
    obj = CustomClass('123')
    # call a method on the custom object
    obj.report_data()
    # await coroutine method
    await obj.sleep_and_report_data()

Tying this together, the complete example below implements this.

# SuperFastPython.com
# example of object method as coroutine
import asyncio

# define a custom class
class CustomClass(object):
    # constructor
    def __init__(self, data):
        # store the data
        self.data = data

    # regular method
    def report_data(self):
        # report the value of data
        print(f'Method: {self.data}')

    # method as coroutine
    async def sleep_and_report_data(self):
        # suspend a moment
        await asyncio.sleep(2)
        # report the data
        print(f'Coroutine: {self.data}')

# main coroutine
async def main():
    # create the custom object
    obj = CustomClass('123')
    # call a method on the custom object
    obj.report_data()
    # await coroutine method
    await obj.sleep_and_report_data()

# start the event loop
asyncio.run(main())

Running the example first creates the main() coroutine and starts the asyncio event loop.

The main() coroutine runs and creates an instance of our CustomClass class and passes in a string as an argument to the constructor.

The class constructor runs and stores the argument in an instance variable.

The main() coroutine then calls the report_data() regular method on the object instance, which reports the value of the instance variable.

Next, the main() coroutine awaits the sleep_and_report_data() coroutine method.

This suspends the main() coroutine and the sleep_and_report_data() runs. It suspends for a moment with a sleep, then resumes and reports the value of the instance variable before terminating.

Finally, the main() coroutine resumes and terminates, closing the event loop.

This highlights how we can define a coroutine method and await it within our asyncio program alongside calling regular object methods.

Method: 123
Coroutine: 123

Next, let's look at the effect of calling a coroutine method outside of an asyncio program.

Example Cannot Call Coroutine Methods Outside of Asyncio

We can explore an example of calling a coroutine method like a regular method outside of an asyncio program.

In this case, we can update the above example to create an instance of our CustomClass class and call both methods directly in a regular Python program.

...
# create the custom object
obj = CustomClass('123')
# call a method on the custom object
obj.report_data()
# await coroutine method
obj.sleep_and_report_data()

We expect this to raise a RuntimeWarning given that a coroutine was created and not executed.

Tying this together, the complete example is listed below.

# SuperFastPython.com
# example of object method as coroutine
import asyncio

# define a custom class
class CustomClass(object):
    # constructor
    def __init__(self, data):
        # store the data
        self.data = data

    # regular method
    def report_data(self):
        # report the value of data
        print(f'Method: {self.data}')

    # method as coroutine
    async def sleep_and_report_data(self):
        # suspend a moment
        await asyncio.sleep(2)
        # report the data
        print(f'Coroutine: {self.data}')

# create the custom object
obj = CustomClass('123')
# call a method on the custom object
obj.report_data()
# await coroutine method
obj.sleep_and_report_data()

Running the example first creates an instance of our CustomClass object and passes it a string argument that is stored in an instance variable of the object.

The report_data() method is then called which then reports the value of the instance variable stored in the object.

Next, the sleep_and_report_data() coroutine method is called. This creates and returns a coroutine object.

The program then terminates. Because the created coroutine object was never executed, a RuntimeWarning is reported.

This highlights that we cannot call a coroutine method like a regular method outside of an asyncio program.

Method: 123
RuntimeWarning: coroutine 'CustomClass.sleep_and_report_data' was never awaited
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

Takeaways

You now know how to define object methods as coroutines.