INW013 async-blocking¶

Co robi¶

Opt-in, a po włączeniu błąd. Zgłasza synchroniczne wywołanie bazy danych, cache'u albo chmury wewnątrz async def, na samym wywołaniu. Liczą się dwa rodzaje wywołań:

• Metoda blokującego odbiorcy. Odbiorca to nazwa, której metody wykonują synchroniczne I/O:
  • parametr, którego adnotacja wskazuje blokujący typ: wprost, Annotated[T, ...], Optional[T], T | None albo alias zdefiniowany w tym samym pliku (DbSession = Annotated[Session, Depends(get_db)]);
  • nazwa, którą moduł albo funkcja przypisuje z wywołania blokującego typu lub fabryki (cache = redis.Redis(), s3 = boto3.client("s3"), with Session(engine) as session) albo opisuje nim w adnotacji (client: Redis = make_client());
  • wywołanie takiego typu wprost w wyrażeniu: Session(engine).scalar(query).
• Wywołanie, które samo blokuje: connect sterownika DB-API.
• Wywołanie synchronicznej funkcji pomocniczej, która robi jedno z powyższych (#294): async def wywołuje zwykłą funkcję def z własnego kodu, w tym samym albo innym module, a jej ciało wykonuje blokujące wywołanie. FastAPI wywołuje taką funkcję bezpośrednio, nie w puli wątków, więc jej ciało działa na pętli zdarzeń. Zgłoszenie stoi na wywołaniu w async def i podaje funkcję pomocniczą, jej plik i linię oraz jej pierwsze blokujące wywołanie.

Wbudowana lista:

extend-blocking-calls i extend-blocking-types dodają własne pozycje. Pliki są sprawdzane niezależnie od warstwy, poza modułami testowymi (przywraca je check-tests = true, zob. tabele opcji); modules zawęża regułę do części z nich, na przykład do routerów.

Dlaczego to źle¶

async def działa na pętli zdarzeń. Kiedy synchroniczne wywołanie czeka na bazę danych, Redisa albo AWS, czeka też pętla, a z nią każde inne żądanie obsługiwane przez ten worker. FastAPI radzi używać async def, chyba że funkcja operacji ścieżki korzysta z kodu, który wykonuje blokujące I/O, a zwykłą funkcję def (trasę albo zależność) uruchamia w puli wątków, żeby nie blokowała serwera (FastAPI: Concurrency and async / await). Agenci kopiują db: Session = Depends(get_db) z tutorialu o SQL i dopisują async def do trasy. Testy przechodzą, bo wysyłają jedno żądanie naraz; spowolnienie wychodzi dopiero pod obciążeniem.

Reguły ASYNC w Ruff już łapią blokujące wywołania HTTP, podprocesy, open, time.sleep, input i pathlib w funkcjach asynchronicznych (ASYNC210 do ASYNC251). Nie znają Session z SQLAlchemy, redis-py, boto3 ani sterowników baz danych, a typ odbiorcy zwykle wynika z adnotacji albo z klienta na poziomie modułu, nie z nazwy samego wywołania. INW013 wypełnia tę lukę, a resztę zostawia Ruffowi.

Przykład¶

[tool.inwards]
layers = [
  { name = "domain", modules = ["store.domain"] },
  { name = "api", modules = ["store.api"] },
]

[tool.inwards.rules]
extend-select = ["INW013"]

from sqlalchemy import create_engine
from sqlalchemy.orm import Session

engine = create_engine("sqlite:///store.db")


def get_db():
    with Session(engine) as session:
        yield session

from typing import Annotated

import redis
from fastapi import APIRouter, Depends
from sqlalchemy import text
from sqlalchemy.orm import Session

from store.api.db import get_db

router = APIRouter()
cache = redis.Redis()


@router.get("/orders/{order_id}")
async def get_order(order_id: int, db: Annotated[Session, Depends(get_db)]) -> dict:
    cached = cache.get(f"order:{order_id}")
    if cached is not None:
        return {"id": order_id, "total": int(cached)}
    total = db.scalar(text("select total from orders where id = :id"), {"id": order_id})
    return {"id": order_id, "total": total}

inwards check store

store/api/orders.py:16:14: INW013 `cache.get` sends a command through a synchronous Redis client inside `async def get_order`. That blocks the event loop, and every other request on this worker, until Redis answers.
  fix: Use `redis.asyncio.Redis` and await the call, or make `get_order` a plain `def`.
    1. Replace `redis.Redis` with `redis.asyncio.Redis`, and `await` the call to `cache.get`.
    2. If `get_order` is a FastAPI or Starlette route or dependency, declare it with `def` instead: FastAPI runs a plain `def` in its threadpool, off the event loop.
    3. If the call has to stay synchronous inside `async def`, run it in a worker thread: `await anyio.to_thread.run_sync(...)`, `await asyncio.to_thread(...)` or Starlette's `run_in_threadpool`.
  docs: https://sircypkowskyy.github.io/inwards/rules/INW013/

store/api/orders.py:19:13: INW013 `db.scalar` talks to the database through a synchronous SQLAlchemy `Session` inside `async def get_order`. That blocks the event loop, and every other request on this worker, until the database answers.
  fix: Use `AsyncSession` from `sqlalchemy.ext.asyncio` and await the call, or make `get_order` a plain `def`.

Poprawione najprościej: trasa jest zwykłą funkcją def, więc FastAPI uruchamia ją w puli wątków. Te same wywołania teraz przechodzą:

from typing import Annotated

import redis
from fastapi import APIRouter, Depends
from sqlalchemy import text
from sqlalchemy.orm import Session

from store.api.db import get_db

router = APIRouter()
cache = redis.Redis()


@router.get("/orders/{order_id}")
def get_order(order_id: int, db: Annotated[Session, Depends(get_db)]) -> dict:
    cached = cache.get(f"order:{order_id}")
    if cached is not None:
        return {"id": order_id, "total": int(cached)}
    total = db.scalar(text("select total from orders where id = :id"), {"id": order_id})
    return {"id": order_id, "total": total}

inwards check store

Druga poprawka zostawia async def i przechodzi na klientów asynchronicznych: redis.asyncio.Redis oraz AsyncSession z create_async_engine, z każdym wywołaniem poprzedzonym await.

Jeden krok w głąb funkcji pomocniczej¶

Trasa jest async def, a praca z bazą danych siedzi w funkcji serwisu:

[tool.inwards.rules]
extend-select = ["INW013"]

from sqlalchemy import insert


def create_user(db, name):
    db.add(User(name=name))
    db.execute(insert(Audit))
    db.commit()
    return name

from typing import Annotated

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session

from shop.application import users

router = APIRouter()


@router.post("/users")
async def create(name: str, db: Annotated[Session, Depends(get_db)]):
    return users.create_user(db, name)

shop/api/routes.py:13:12: INW013 `users.create_user` runs the plain `def create_user` (shop/application/users.py, line 4) on the event loop inside `async def create`, and `db.execute` on line 6 of it, the first of 2 blocking calls, talks to the database through a synchronous SQLAlchemy `Session`. That blocks the event loop, and every other request on this worker, until the database answers.
  fix: Run `create_user` in a worker thread, make it `async def` with `AsyncSession` from `sqlalchemy.ext.asyncio`, or make `create` a plain `def`.

create_user nie ma adnotacji przy db; trasa przekazuje własne db, czyli Session, więc db w funkcji pomocniczej też nim jest. Przekazanie funkcji do wątku roboczego przechodzi: await run_in_threadpool(users.create_user, db, name) niczego nie wywołuje na pętli.

Jak działa¶

Najpierw działa filtr tekstowy: plik bez async def albo taki, który nie wymienia żadnej z nazw sqlalchemy, sqlmodel, redis, boto3, sqlite3, psycopg, pymysql, MySQLdb i mysql ani pierwszego członu skonfigurowanego wzorca, nigdy nie jest parsowany. Plik, który przejdzie filtr, jest parsowany raz, na potrzeby tej reguły.

• Funkcje: każde async def, łącznie z metodami i funkcjami zagnieżdżonymi. Liczy się jego ciało aż do zagnieżdżonego def, lambda albo class: ich ciała nie wykonują się, gdy funkcja działa, dlatego await anyio.to_thread.run_sync(lambda: db.execute(q)) przechodzi.
• Nazwy są kwalifikowane przez importy pliku, więc from redis import Redis as Cache i cache: Cache nadal znaczą redis.Redis.
• Odbiorcy pochodzą z przypisań na poziomie modułu, potem z parametrów funkcji (parametr przesłania nazwę modułu o tej samej pisowni), a na końcu z nazw przypisanych w jej ciele. Blokujące wywołanie poprzedzone await też jest zgłaszane: klient synchroniczny nie staje się asynchroniczny od await.
• Jeden krok (#294). Dla każdego wywołania w async def, które samo nie jest blokujące, Inwards szuka funkcji, którą ono wskazuje: funkcji z najwyższego poziomu tego samego modułu wywołanej gołą nazwą albo, przez indeks modułów projektu, funkcji z najwyższego poziomu innego modułu z własnego kodu, idąc za reeksportami w __init__.py pakietu. Wchodzi tylko do zwykłego def bez dekoratora: funkcja pomocnicza async def ma własne sprawdzenie, a dekorator może przenieść wywołanie poza pętlę. Wywołanie poprzedzone await i wywołanie w zagnieżdżonym def albo lambda nie są śledzone, a funkcja przekazana do run_in_threadpool, anyio.to_thread.run_sync, asyncio.to_thread albo loop.run_in_executor w ogóle nie jest wywoływana. Odbiorcy funkcji pomocniczej to odbiorcy jej modułu, jej parametry z adnotacją, nazwy przypisane w jej ciele, a dla parametru bez adnotacji odbiorca, którego wywołujący przekazuje gołą nazwą, pozycyjnie albo przez słowo kluczowe. Wywołania samej funkcji pomocniczej nie są śledzone dalej.
• Koszt. Każdy moduł odczytany w tym kroku jest parsowany i czytany raz na uruchomienie, niezależnie od tego, ile sprawdzanych plików do niego woła (#359), i tylko wtedy, gdy async def wywołuje w nim funkcję; wywołanie metody obiektu z poziomu modułu (service.create(...)) niczego nie czyta, a sprawdzany plik, który krok już sparsował, nie jest parsowany ponownie; follow-modules ogranicza, które moduły są czytane. Gdy śledzenie jest włączone, sprawdzany plik z async def może dojść do blokującej funkcji pomocniczej w module, który wymienia bibliotekę, choć sam plik jej nie wymienia. W uruchomieniu z wieloma takimi plikami (co najmniej 200 i co dwudziesty plik uruchomienia) drugi filtr czyta raz każdy moduł z listingu, parsuje tylko te, których importy mogą złożyć nazwę blokującego typu, i pomija plik, który nie wymienia ani blokującego typu, ani żadnego z ich zwykłych def, które blokują, pod własną nazwą ani pod aliasem nadanym przez reeksport (#369). Mniejsze uruchomienie, takie jak w hooku, parsuje każdy taki plik. Przy follow-modules = [] znów działa opisany wyżej filtr tekstowy.
• Jedno zgłoszenie na wywołanie. sqlite3.connect(p).execute(q) jest zgłaszane raz, na sqlite3.connect; db.query(Order).filter(...).all() raz, na db.query.

Jak naprawić¶

1. Przejdź na klienta asynchronicznego tej biblioteki i dodaj await przed wywołaniem: AsyncSession i create_async_engine z sqlalchemy.ext.asyncio, redis.asyncio.Redis, asyncpg albo psycopg.AsyncConnection, aiosqlite, aiomysql, aioboto3.
2. Albo, w przypadku trasy lub zależności FastAPI czy Starlette, zadeklaruj ją jako def: framework uruchomi ją w puli wątków.
3. Albo zostaw wywołanie synchroniczne i uruchom je w wątku roboczym: await anyio.to_thread.run_sync(...), await asyncio.to_thread(...) albo run_in_threadpool.

Bezpieczeństwo poprawki¶

Bez automatycznej poprawki (autofix: false). Przejście na klienta asynchronicznego zmienia zależność, fabrykę sesji i każde miejsce wywołania; przejście na def zmienia sposób, w jaki trasa się wykonuje.

Konfiguracja¶

Poziom error. Opt-in: włącz ją przez extend-select. Preset fastapi polecenia inwards init włącza INW013 jako ostrzeżenie we wszystkich modułach. Opcje są w [tool.inwards.rules.async-blocking] (tabele opcji):

[tool.inwards.rules]
extend-select = ["INW013"]

[tool.inwards.rules.async-blocking]
modules = ["shop.*.router", "shop.*.service"]   # optional: only these modules
extend-blocking-calls = ["shop.legacy.fetch_*"] # calls that block by themselves
extend-blocking-types = ["pymongo.MongoClient"] # every method of these blocks
follow-modules = ["shop.*.service"]             # optional: the modules a hop reads

• extend-blocking-calls: kwalifikowane nazwy, z * i ? jako wildcardami, funkcji, których wywołanie blokuje. Dochodzą do wbudowanych sterowników.
• extend-blocking-types: kwalifikowane nazwy, z * i ? jako wildcardami, typów albo fabryk, których każda metoda blokuje. Parametr z taką adnotacją albo nazwa przypisana z wywołania takiego typu jest odbiorcą.
• follow-modules: prefiksy modułów albo selektory, zapisane jak w modules warstwy, wskazujące moduły z własnego kodu, do których może prowadzić krok. Pakiety nad modułem pasującym do wpisu też są czytane, bo reeksport może przechodzić przez ich __init__.py. Bez tej opcji krok może prowadzić do każdego modułu z własnego kodu; [] ogranicza sprawdzenie do funkcji pomocniczych ze sprawdzanego modułu. Funkcje pomocnicze ze sprawdzanego modułu są śledzone zawsze.

Wbudowanej listy nie da się skrócić; użyj modules albo wyciszenia w linii dla pojedynczego wywołania: # inwards: ignore[INW013] reason="..." w linii wywołania.

Szablon może włączyć INW013 dla jednej ze swoich ról: router = { async-blocking = true } ustawia modules na moduły roli.

Znane ograniczenia¶

• Tylko jeden krok. Synchroniczna funkcja pomocnicza, która wywołuje kolejną blokującą funkcję, nie jest zgłaszana; podobnie metoda (service.create(db) na instancji, self.repo.save()), metoda statyczna klasy albo funkcja zapisana w zmiennej. Opakuj wewnętrzne wywołanie albo wycisz je z powodem.
• Co dociera do funkcji pomocniczej. Do jej parametru trafia tylko odbiorca przekazany gołą nazwą; helper(db=get_session()), helper(*args) albo odbiorca w słowniku nic nie mówią. Funkcja pomocnicza z adnotacją przy parametrze nie potrzebuje niczego od wywołującego.
• Pośredniość. self.db, klient przekazany przez **kwargs albo słownik, klient zbudowany przez fabrykę z innego modułu (SessionLocal() z sessionmaker w db.py) i alias typu zaimportowany z innego modułu albo zapisany jako napis ukrywają odbiorcę. Dodaj adnotację do parametru.
• Ponowne przypisanie. Nazwa zostaje odbiorcą, gdy raz stanie się nim przez przypisanie: po db = Session(engine), a potem db = AsyncSession(engine), db nadal liczy się jako blokujące.
• Parametry bez adnotacji nie są odbiorcami: async def f(db) nic nie mówi o db.
• Nie każda metoda blokuje. Rodziny Engine, redis-py i sterowników liczą każdą metodę poza kilkoma; rzadka metoda działająca tylko lokalnie da fałszywe zgłoszenie.
• Kod startowy. Asynchroniczny lifespan albo hook startowy, który raz otwiera połączenie, raz blokuje pętlę. Zawęź modules albo wycisz linię z powodem.
• Zmierzona na korpusie (#182, rozdział 6): 1 zgłoszenie w pięciu repozytoriach, synchroniczne Session.query w async def handlera zapytania w python-ddd, i jest trafne. Kod asynchroniczny Polara używa AsyncSession i nie dostaje żadnego. Na Polarze zimne pełne sprawdzenie z włączonym INW013 trwa 0,73 s wobec 0,55 s bez niego (#369); follow-modules zawęża je dalej.

Źródła¶

• #293: reguła; #98: reguły treści w zakresie roli, z których pochodzi.
• FastAPI: Concurrency and async / await: trasy i zależności def działają w puli wątków, async def na pętli zdarzeń.
• SQLAlchemy: Asynchronous I/O (asyncio): metody awaitable w AsyncSession.
• redis-py: Asyncio examples: redis.asyncio i pipeline'y, które niczego nie wysyłają przed execute().
• FastAPI: Concurrency and async / await, "Other utility functions": zwykłą funkcję def, którą wywołujesz sam, FastAPI wywołuje bezpośrednio, nie w puli wątków.
• asyncio.to_thread, AnyIO: Working with threads i run_in_threadpool ze Starlette: każda przyjmuje funkcję i jej argumenty, więc wywołanie działa w wątku roboczym.
• Reguły flake8-async w Ruff: blokujące wywołania, które Ruff już zgłasza.