You can identify common problems in asyncio programs using popular Python static analysis tools such as pylint and flake8.
In this tutorial, you will discover how to lint python code for common asyncio problems.
Let’s get started.
What is Lint or Linting?
Linting refers to tools for performing a static analysis check of code.
Lint is the computer science term for a static code analysis tool used to flag programming errors, bugs, stylistic errors and suspicious constructs.
— Lint (software), Wikipedia.
It is a helpful check of Python code to check for common programming errors that can be performed regularly and automatically, such as each time new code is committed as part of a continuous integration system.
In computer science, static program analysis (or static analysis) is the analysis of computer programs performed without executing them …
— Static program analysis, Wikipedia.
Linting Python code involves using a tool (a linter) to analyze and check the source code for adherence to coding standards, style guidelines, and potential programming errors. Linting tools review the code without executing it, highlighting issues such as syntax errors, coding style violations, and potential bugs.
Linting is essential for several reasons:
- Consistency: Linting enforces coding standards and style guidelines, ensuring that code across a project or team follows a consistent and readable format. This improves code maintainability and readability.
- Error Prevention: Linters identify potential bugs, such as syntax errors, undefined variables, and other common mistakes. By catching these issues early in the development process, developers can prevent runtime errors and improve code reliability.
- Coding Standards: Linters enforce coding standards and best practices. They help developers write code that is in line with community-accepted conventions, which makes it easier for others to understand and contribute to the codebase.
Linting Python code is a vital practice that promotes code consistency, identifies potential errors, enforces coding standards and style guidelines, and ultimately leads to more readable, reliable, and maintainable code.
Next, let’s consider linting asyncio programs.
Run loops using all CPUs, download your FREE book to learn how.
Need to Lint For Common Asyncio Errors
We can use static analysis to check for common errors in asyncio programs.
There are three common errors that we may see in an asyncio program, they are:
- Never retrieved exceptions.
- Never awaited coroutine.
- Calls that block the event loop.
Generally, these are runtime errors.
The first two, never retrieved exceptions and never awaited coroutines typically result in a RuntimeWarning, and therefore are often easy to identify and repair.
For example, you can learn more about never-retrieved exceptions in the tutorial:
You can learn more about never-awaited coroutines in the tutorial:
The last error does not result in a runtime warning, but we may be able to identify examples of calls that block the event loop using static analysis.
There are also other errors that we may be able to identify. These are errors that are also likely to result in a syntax error, causing the program to fail to execute.
Such as:
- Using an await expression outside of a coroutine.
- Using an async context manager outside of a coroutine.
- Using an async generator outside of a coroutine.
Next, let’s explore how we might lint our asyncio programs.
How to Lint Asyncio Programs
Perhaps two of the most common Python linters are:
- Pylint
- Flake8
Let’s take a closer look at asyncio support for each.
How to Use Pylint For Asyncio Programs
Pylint is a popular and simple-to-use static analysis tool for Python.
Pylint analyses your code without actually running it. It checks for errors, enforces a coding standard, looks for code smells, and can make suggestions about how the code could be refactored.
— Pylint, GitHub.
We can install Pylint using our favorite Python package manager, such as pip.
For example:
1 |
pip install pylint |
We can then apply pylint to a project or a file by invoking it directly.
For example:
1 |
pylint program.py |
It will then identify issues identified with the program.
By default, pylint can identify three types of errors in asyncio programs, they are:
These are reasonable, but I would argue that running the program will cause a syntax error and identify these issues for us.
Helpfully, there is a plugin for pylint called “pylint-blocking-calls” that can be used to identify blocking calls made in the event loop.
This is a pylint plugin that checks for blocking calls in your async python code.
— Pylint Blocking Calls Checker, GitHub.
We can install this plugin using our favorite package manager, such as pip.
For example:
1 |
pip install pylint-blocking-calls |
Using the plugin is cumbersome.
We must define a variable named BLOCKING_FUNCTION_NAMES that specifies all of the locking functions to look for.
This can be achieved using an environment variable or programmatically.
For example:
1 |
BLOCKING_FUNCTION_NAMES="^time.sleep$" |
We then must start pylint and specify to load the plugin via the flag:
1 |
--load-plugins=pylint_blocking_calls |
A call to check a program for calls to the time.sleep() blocking function may look like:
1 |
BLOCKING_FUNCTION_NAMES="^time.sleep$" pylint --load-plugins=pylint_blocking_calls program.py |
It might be helpful if someone comes along and defines a long list of offending blocking calls in the standard library.
How to Use Flake8 For Asyncio Programs
Flake8 is another widely used static analysis tool for Python.
flake8 is a python tool that glues together pycodestyle, pyflakes, mccabe, and third-party plugins to check the style and quality of some python code.
— flake8, GitHub.
Many other tools build on top of the flake8 ecosystem (the rules specifically), such as the popular ruff linter.
We can install Flake8 using our favorite Python package manager, such as pip.
For example:
1 |
pip install flake8 |
We can then apply flake8 to a project or a file by invoking it directly.
For example:
1 |
flake8 program.py |
It will then identify issues identified with the program.
By default, it does not appear that flake8 has rules to identify problems specific to asyncio programs.
Nevertheless, there is a plugin named flake8-async that we can install that provides async-specific rules.
A flake8 plugin that checks for bad async / asyncio practices.
— flake8-async, GitHub.
We can install the plugin using our favorite package manager, such as pip.
For example:
1 |
pip install flake8-async |
The plugin provides 3 asyncio rules, they are:
- ASYNC100: Warning about the use of a blocking http call inside an async def
- ASYNC101: Warning about the use of open, time.sleep or methods in subprocess, inside an async def.
- ASYNC102: Warning about the use of unsafe methods in os inside an async def.
Using flake8 with the plugin on an asyncio program is straightforward.
For example:
1 |
flake8 program.py |
It will report all issues, including asyncio rules directly.
Now that we know how to lint Python code for asyncio issues, let’s look at some worked examples.
Free Python Asyncio Course
Download your FREE Asyncio PDF cheat sheet and get BONUS access to my free 7-day crash course on the Asyncio API.
Discover how to use the Python asyncio module including how to define, create, and run new coroutines and how to use non-blocking I/O.
Examples of Pylint for Asyncio Code
In this section, we will explore some common errors in asyncio programs that can be identified automatically via pylint.
Specifically, we will at an example of :
- Pylint reporting on the use of an “await” expression outside of a coroutine.
- Pylint reporting on a blocking call made in the event loop.
Pylint Async Outside of Coroutine
We can develop a simple asyncio program that uses the await expression outside of a coroutine.
In this example, the program will define a task that sleeps for a moment, runs this task in the background, sleeps in the main coroutine, and then reports a done message.
The error will be that we will forget to define the work() task as a coroutine. Instead, we will define it as a Python function.
This will cause the program to not run with a syntax error, but it is also an error we can detect before running via pylint.
Firstly, we can define the work() task that awaits a call to sleep but is not defined as a coroutine.
1 2 3 4 |
# task that does work def work(): # block for a moment await asyncio.sleep(1) |
Note, this should be defined using the “async def” expression:
1 2 3 4 |
# task that does work async def work(): # block for a moment await asyncio.sleep(1) |
Next, we can define the main() coroutine that will report a message, run the work() task in the background, sleep a moment, and then report a done message.
1 2 3 4 5 6 7 8 9 10 |
# main coroutine async def main(): # report a message print('Starting') # run the task _ = asyncio.create_task(work()) # block for a moment await asyncio.sleep(2) # report a message print('Done') |
Finally, we can start the event loop.
1 2 3 |
... # start the event loop asyncio.run(main()) |
Tying this together, the complete example is listed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
# SuperFastPython.com # example of await outside of coroutine import asyncio # task that does work def work(): # block for a moment await asyncio.sleep(1) # main coroutine async def main(): # report a message print('Starting') # run the task _ = asyncio.create_task(work()) # block for a moment await asyncio.sleep(2) # report a message print('Done') # start the event loop asyncio.run(main()) |
Running the program results in a syntax error.
1 |
SyntaxError: 'await' outside async function |
Now we can lint the program using pylint.
1 |
pylint program.py |
This results in one issue reported.
Specifically, we have used an “await” expression outside of a coroutine, in this case, our work() function.
This highlights how we can detect syntax errors in asyncio programs using pylint.
1 2 3 4 5 |
************* Module program program.py:8:4: E1142: 'await' should be used within an async function (await-outside-async) ------------------------------------------------------------------ Your code has been rated at 4.44/10 (previous run: 4.44/10, +0.00) |
Next, let’s look at how we might detect a blocking call in an asyncio program using pylint.
Pylint Event Loop Blocking Call
We can explore an example of pylint identifying a call that blocks the asyncio event loop.
In this case, we will define a coroutine that calls the time.sleep() function.
This will block the event loop for the duration of the sleep, a bad practice.
We can identify this issue using the pylint_blocking_calls pylint plugin and explicitly check for calls to the time.sleep() function.
We can define the work() coroutine that will make the call to time.sleep().
1 2 3 4 |
# task that does work async def work(): # block for a moment time.sleep(1) |
The complete example is listed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 |
# SuperFastPython.com # example of a call that blocks the event loop import time import asyncio # task that does work async def work(): # block for a moment time.sleep(1) # main coroutine async def main(): # report a message print('Starting') # run the task _ = asyncio.create_task(work()) # block for a moment await asyncio.sleep(2) # report a message print('Done') # start asyncio.run(main()) |
Running the program results in a normal execution with two messages reported.
1 2 |
Starting Done |
We can now call pylint with the plugin and check for calls to time.sleep within coroutines.
1 |
BLOCKING_FUNCTION_NAMES="^time.sleep$" pylint --load-plugins=pylint_blocking_calls program.py |
This results in one issue identified, specifically that the time.sleep() blocking call was made within a coroutine.
This highlights how we can explicitly identify blocking calls in coroutines, although we must know what the blocking calls will be beforehand so that we can search for them.
1 2 3 4 5 |
************* Module program program.py:9:4: W0002: time.sleep (blocking-call) ------------------------------------------------------------------ Your code has been rated at 9.00/10 (previous run: 9.00/10, +0.00) |
Next, let’s look at examples of linting with flake8.
Overwhelmed by the python concurrency APIs?
Find relief, download my FREE Python Concurrency Mind Maps
Examples of Flake8 for Asyncio Code
In this section, we will explore some common errors in asyncio programs that can be identified automatically via Flake8.
Specifically, we will at an example of:
- Flake8 reporting on opening a file in a coroutine.
- Flake8 reporting on a blocking call made in the event loop.
Flake8 Opening File in Coroutine
We can explore an example of flake8 identifying an issue with opening a file within a coroutine.
In this case, we can use the open() built-in function that is a blocking call and create a new file. Checking this file with flake8 will identify an issue.
We would prefer to perform blocking file I/O tasks in a separate worker thread that can be awaited.
Firstly, we can define a main() coroutine that opens the file and reports a done message.
1 2 3 4 5 6 |
# main coroutine async def main(): # open a file open('test.txt', 'w') # report a message print('Done') |
We can then start the event loop and run the coroutine.
1 2 3 |
... # start the event loop asyncio.run(main()) |
Tying this together, the complete example is listed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 |
# SuperFastPython.com # example of opening a file in a coroutine import asyncio # main coroutine async def main(): # open a file open('test.txt', 'w') # report a message print('Done') # start asyncio.run(main()) |
Running the program runs normally, creating the file and reporting a done message.
1 |
Done |
Now we can check the program using flake8 with the plugin installed.
1 |
flake8 program.py |
This identifies one issue specifically the blocking file I/O operation is being performed in a coroutine.
This highlights how we can automatically discover and report on file IO tasks in asyncio programs.
1 |
program.py:8:5: ASYNC101: blocking sync call in async function, use framework equivalent |
Next, let’s look at an example of an explicit blocking call in a coroutine.
Flake8 Event Loop Blocking Call
We can explore an example of Flake8 identifying a call that blocks the asyncio event loop.
In this case, we will define a coroutine that calls the time.sleep() function.
This will block the event loop for the duration of the sleep, a bad practice.
We can identify this issue using the pylint_blocking_calls Flake8 plugin and explicitly check for calls to the time.sleep() function.
We can define the work() coroutine that will make the call to time.sleep().
1 2 3 4 |
# task that does work async def work(): # block for a moment time.sleep(1) |
The complete example is listed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 |
# SuperFastPython.com # example of a call that blocks the event loop import time import asyncio # task that does work async def work(): # block for a moment time.sleep(1) # main coroutine async def main(): # report a message print('Starting') # run the task _ = asyncio.create_task(work()) # block for a moment await asyncio.sleep(2) # report a message print('Done') # start asyncio.run(main()) |
Running the program results in a normal execution with two messages reported.
1 2 |
Starting Done |
We can now call Flake8 with the plugin installed.
1 |
flake8 program.py |
This results in one issue identified, specifically the time.sleep() blocking call was made within a coroutine.
This highlights how we can explicitly identify blocking calls in coroutines using Flake8.
1 |
program.py:9:5: ASYNC101: blocking sync call in async function, use framework equivalent |
Further Reading
This section provides additional resources that you may find helpful.
Python Asyncio Books
- Python Asyncio Mastery, Jason Brownlee (my book!)
- Python Asyncio Jump-Start, Jason Brownlee.
- Python Asyncio Interview Questions, Jason Brownlee.
- Asyncio Module API Cheat Sheet
I also recommend the following books:
- Python Concurrency with asyncio, Matthew Fowler, 2022.
- Using Asyncio in Python, Caleb Hattingh, 2020.
- asyncio Recipes, Mohamed Mustapha Tahrioui, 2019.
Guides
APIs
- asyncio — Asynchronous I/O
- Asyncio Coroutines and Tasks
- Asyncio Streams
- Asyncio Subprocesses
- Asyncio Queues
- Asyncio Synchronization Primitives
References
Takeaways
You now know how to lint Python code for common asyncio problems.
Did I make a mistake? See a typo?
I’m a simple humble human. Correct me, please!
Do you have any additional tips?
I’d love to hear about them!
Do you have any questions?
Ask your questions in the comments below and I will do my best to answer.
Photo by Luke Bender on Unsplash
Do you have any questions?