Problem Statement & Motivation

The Core Problem

Blocking function calls inside Python async contexts silently destroy concurrency. When an async def function calls a blocking operation – such as time.sleep(), requests.get(), or any synchronous I/O – the entire event loop freezes. No other coroutines can execute until the blocking call completes. The application appears to work correctly in isolation but fails catastrophically under load.

This is an insidious class of bug because:

1. The code runs without errors. Python does not raise exceptions or warnings when blocking calls occur in async contexts.
2. Tests pass in isolation. A single request completes successfully, masking the concurrency failure.
3. Production failures are mysterious. Under concurrent load, the application becomes unresponsive, timeouts cascade, and the root cause is non-obvious.
4. The bug propagates transitively. A blocking call buried five levels deep in the call stack poisons every async caller above it.

Why Existing Tools Fail

Current linters detect only direct blocking calls. They cannot trace blocking behavior through call chains.

Example from the Strato README:

import time

async def handler():
    baz()  # Looks innocent – no linter flags this

def baz():
    time.sleep(1)  # The actual blocking call

• flake8-async and ruff ASYNC2XX flag time.sleep(1) if it appears directly in an async def, but they do not flag baz() when called from handler().
• PyCG builds call graphs but does not understand async semantics or blocking behavior.

The result: developers must manually audit every function in the call chain to determine if it eventually blocks. This is infeasible in large codebases.

Detection Case Matrix

Tool Comparison

Motivating Examples

Indirect Blocking (README Example)

import time

async def handler():
    baz()  # No existing tool flags this line

def baz():
    time.sleep(1)  # Blocks the event loop

Strato output:

STRATO002: Transitive blocking call reachable from async context
  --> example.py:4:5
   |
 4 |     baz()
   |     ^^^^^ blocking call here
   |
   = note: call chain: handler -> baz -> time.sleep (length: 3)

Blocking Property

import requests

class DataFetcher:
    @property
    def data(self):
        return requests.get("https://api.example.com/data").json()

async def process():
    fetcher = DataFetcher()
    result = fetcher.data  # Looks like attribute access, actually blocks

Problem: The property getter performs synchronous HTTP I/O. The access fetcher.data appears to be a simple attribute read but blocks the event loop.

Strato output:

STRATO003: Async function 'process' accesses blocking property 'DataFetcher.data'
  --> example.py:10:14
   |
10 |     result = fetcher.data
   |              ^^^^^^^^^^^^ blocking property access
   |
   = note: property getter calls requests.get (blocking)

Blocking Dunder Method

import requests

class RemoteObject:
    def __str__(self):
        return requests.get("https://api.example.com/status").text

async def log_status(obj):
    print(str(obj))  # Implicit __str__ call blocks

Problem: The __str__ method performs network I/O. The str(obj) call is implicit and blocks the event loop.

Strato output:

STRATO004: Async function 'log_status' calls blocking dunder method 'RemoteObject.__str__'
  --> example.py:8:11
   |
 8 |     print(str(obj))
   |           ^^^^^^^^ blocking dunder call
   |
   = note: __str__ method calls requests.get (blocking)

Cross-File Blocking

utils.py:

import time

def slow_util():
    time.sleep(2)

main.py:

from utils import slow_util

async def handler():
    slow_util()  # Blocking call hidden in another file

Strato output:

STRATO002: Transitive blocking call reachable from async context
  --> main.py:4:5
   |
 4 |     slow_util()
   |     ^^^^^^^^^^^ blocking call here
   |
   = note: call chain: handler -> slow_util -> time.sleep (length: 3)
   = note: slow_util defined in utils.py:3

Deep Transitive Chain

import time

async def handler():
    level_1()

def level_1():
    level_2()

def level_2():
    level_3()

def level_3():
    time.sleep(1)

Problem: The blocking call is four levels deep. Manual inspection requires tracing through multiple function definitions.

Strato output:

STRATO002: Transitive blocking call reachable from async context
  --> example.py:4:5
   |
 4 |     level_1()
   |     ^^^^^^^^^ blocking call here
   |
   = note: call chain: handler -> level_1 -> level_2 -> level_3 -> time.sleep (length: 5)