Running async spatial tests with pytest-asyncio

10 min read

Running async spatial tests with pytest-asyncio is the problem of evaluating many spatial relationships concurrently — querying a PostGIS pool with asyncpg, fanning out per-feature topology checks, validating raster/vector pairs in parallel — without the event loop conflicts, fixture-scope leaks, and CPU-blocking stalls that make a green test suite flaky on the next CI run. The exact tools are pytest-asyncio (≥1.0), the asyncio event loop, an async database driver such as asyncpg, and the asyncio.to_thread offload primitive that keeps synchronous GEOS/GDAL calls off the loop. This page sits beneath topology rule enforcement, the relationship-validation layer of the Spatial Test Pattern Design & Implementation discipline, and shows the precise configuration and code that turns concurrent topology validation into a deterministic, copy-button-ready gate.

Concurrency is not a performance luxury here. A coverage check over a parcel layer is an all-pairs relationship problem; running each pair’s intersects/overlaps predicate sequentially against a synchronous cursor leaves the runner idle on I/O. Async lets the database round-trips overlap while the CPU-bound geometry work is dispatched to a thread pool — but only if the loop is never blocked and every fixture is torn down on the loop that created it.

Root cause: why async spatial tests destabilise CI

Async spatial failures are rarely about the spatial logic. They come from three structural mismatches between how pytest-asyncio manages loops and how spatial libraries behave.

  • Event-loop nesting and policy conflicts. pytest-asyncio creates an isolated event loop per test function by default. Spatial libraries that wrap C extensions — geopandas, shapely, rasterio, the GDAL bindings — may spawn worker threads or hold references to a loop captured at import time. Calling back into the loop from that foreign context raises RuntimeError: This event loop is already running or asyncio.InvalidStateError. The failure surfaces intermittently because it depends on thread scheduling, which is exactly what makes it look like a spatial bug.
  • Fixture scope outliving its loop. An async fixture (@pytest_asyncio.fixture) scoped to session or module opens an asyncpg pool on one loop, but per-function tests run on fresh loops. The pool’s connections are now bound to a dead loop, so the next acquire() either hangs or raises got Future attached to a different loop. Connection exhaustion during parallel topology validation is the same bug wearing a different mask.
  • Blocking the loop with CPU-bound geometry. shapely validity tests, geopandas.overlay, and pyproj transforms are synchronous and CPU-bound. Awaiting them directly inside an async def does not yield — it freezes the entire loop, starving every other coroutine until the call returns. Under pytest-timeout this trips the deadline and masks the real spatial result behind a timeout.

The unifying fix is a strict loop boundary: pin the loop scope, dispatch every synchronous spatial call through asyncio.to_thread, and tear pools down on the loop that created them.

Offloaded async path keeps the event loop free; an inline geometry call freezes it A two-lane comparison. The top lane traces the offloaded flow — coroutine, awaited asyncpg fetch that yields the loop, asyncio.to_thread fan-out into a thread pool of CPU-bound spatial checks, then asyncio.gather and assert — followed by an event-loop timeline whose heartbeat ticks continue unbroken (5/5). The bottom lane traces the inline flow — coroutine, an inline geom.is_valid that does not yield, an event-loop-blocked region — followed by a timeline whose heartbeat ticks stop (0/5) as soon as the synchronous call begins. Offloaded path — the event loop stays free async test coroutine await asyncpg.fetch I/O · loop yields asyncio.to_thread fan-out × N thread pool · w workers shapely.is_valid (CPU) ST_overlaps check (CPU) … feature N (CPU) asyncio.gather collect · assert event loop: free · 5/5 Awaited I/O plus the thread offload interleave the coroutines — the heartbeat probe ticks 5/5. Inline path — one synchronous call freezes the loop async test coroutine await geom.is_valid inline · no yield EVENT LOOP BLOCKED every other coroutine starves event loop: stalled · 0/5 The synchronous call never yields — once it begins (dashed line) the heartbeat probe stops at 0/5. Blocking the event loop with GEOS, and the fix Two timelines. In the blocking timeline a single GEOS union occupies the event loop for nine seconds; queued database queries, timer callbacks and cancellation handling are all shown stacked behind it, and a two-second timeout is marked as firing late at nine seconds because the loop could not run the timer. In the offloaded timeline the GEOS call is dispatched to a worker thread where it releases the global interpreter lock; the event loop timeline stays free, database queries interleave normally, and the two-second timeout fires on schedule and cancels the work. Awaited on the loop — everything queues behind GEOS loop GEOS union · 9 s, no await points inside queued db query timer callback cancellation handling timeout should fire at 2 s actually fires at 9 s Offloaded to a thread — the loop stays responsive loop db query timer db query timer thread GEOS union · GIL released for the duration timeout fires at 2 s and cancels, as written GEOS calls release the interpreter lock, which is precisely what makes a thread the right offload target here rather than a process.

Configuration and offload reference

The settings below are the whole contract. asyncio_mode = "strict" forces an explicit @pytest.mark.asyncio on every coroutine test, so a fixture can never be silently awaited on the wrong loop; loop_scope keeps a session pool and its tests on one loop.

Key / callable Where Value / signature Role in the gate
asyncio_mode pyproject.toml "strict" Require explicit @pytest.mark.asyncio; no auto-marking
@pytest.mark.asyncio(loop_scope=...) test / fixture "session" | "module" | "function" Bind tests and their async fixtures to one loop
@pytest_asyncio.fixture(loop_scope=...) conftest.py matches the pool’s lifetime Keep the pool on the loop that created it
asyncio.to_thread(fn, *args) test body Coroutine Offload CPU-bound Shapely/GDAL off the loop
asyncio.gather(*tasks) test body Future[list] Fan-out concurrent predicate evaluation
pytest-xdist --dist CLI loadfile One loop + one GDAL cache per worker
@pytest.mark.timeout(n) test seconds (pytest-timeout) Per-test deadline; pytest-asyncio has no timeout=

The reason offloading is mandatory is a throughput argument, not a style preference. For NN features each costing tcput_{\text{cpu}} of synchronous geometry work plus tiot_{\text{io}} of database wait, a loop-blocking run costs roughly

Tblock=N (tcpu+tio),T_{\text{block}} = N \,(t_{\text{cpu}} + t_{\text{io}}),

because nothing overlaps. Dispatching the CPU work to a pool of ww threads while I/O is awaited concurrently bounds it near

Tasync≈max⁡ ⁣(N tcpuw,  N tio),T_{\text{async}} \approx \max\!\left(\frac{N\, t_{\text{cpu}}}{w},\; N\, t_{\text{io}}\right),

so the win scales with the worker count only while the loop itself stays free. One inline shapely call collapses the second formula back into the first.

Step-by-step implementation

Step 1 — Pin strict mode and markers

Configure pytest-asyncio 1.0+ in pyproject.toml. strict mode is what prevents an async fixture from being pulled onto a per-function loop by accident.

[tool.pytest.ini_options]
asyncio_mode = "strict"
addopts = "-v --tb=short --strict-markers"
timeout = 120  # global default from pytest-timeout; per-test override with @pytest.mark.timeout
markers = [
    "spatial_async: async spatial validation (deselect with '-m \"not spatial_async\"')",
]

Step 2 — Scope the database pool to its loop

Give the asyncpg pool an explicit loop_scope that matches the tests consuming it, and close it in the same fixture so connections never outlive their loop.

# conftest.py — pytest-asyncio >= 1.0, asyncpg >= 0.29
import asyncpg
import pytest_asyncio

@pytest_asyncio.fixture(loop_scope="session", scope="session")
async def async_db_pool():
    """Session pool pinned to the session loop; torn down on that same loop."""
    # asyncpg takes a plain libpq DSN — not the SQLAlchemy "postgresql+asyncpg" form
    pool = await asyncpg.create_pool(
        dsn="postgresql://user:pass@localhost/spatial_db",
        min_size=2,
        max_size=10,
    )
    yield pool
    await pool.close()  # closes on the loop that opened it — no cross-loop Future

Step 3 — Offload CPU-bound geometry to a thread

Never await a Shapely predicate inline. is_valid is a property, so wrap it in a callable — to_thread needs a function, not its already-evaluated result.

import asyncio
import pytest
from shapely.geometry import shape

@pytest.mark.asyncio(loop_scope="session")
async def test_async_geometry_validation():
    geojson = {"type": "Polygon",
               "coordinates": [[[0, 0], [1, 0], [1, 1], [0, 1], [0, 0]]]}
    geom = shape(geojson)
    # property -> wrap in lambda so a callable (not a bool) is dispatched off-loop
    is_valid = await asyncio.to_thread(lambda: geom.is_valid)
    assert is_valid, "Geometry failed validation"

Step 4 — Fan out concurrent topology rule checks

This is the payoff for topology rule enforcement: fetch candidates with one async round-trip, then evaluate every relationship predicate concurrently in the thread pool. The per-feature function is pure and synchronous so it is trivially poolable.

import asyncio
import pytest
from shapely.wkb import loads as wkb_loads

def _check_overlap_rule(feature_id: int, geom_wkb: bytes) -> dict:
    """CPU-bound relationship check; runs in a thread executor, never on the loop."""
    geom = wkb_loads(geom_wkb)
    # Real logic compares against a prepared spatial index; placeholder validity here.
    return {"id": feature_id, "violates": not geom.is_valid}

@pytest.mark.asyncio(loop_scope="session")
async def test_async_topology_overlap_check(async_db_pool):
    async with async_db_pool.acquire() as conn:
        rows = await conn.fetch(
            "SELECT id, ST_AsEWKB(geom) AS geom "
            "FROM parcels WHERE status = 'pending'"
        )
    tasks = [
        asyncio.to_thread(_check_overlap_rule, r["id"], bytes(r["geom"]))
        for r in rows
    ]
    results = await asyncio.gather(*tasks)
    violations = [r for r in results if r["violates"]]
    assert not violations, f"Topology violations detected: {violations}"

Step 5 — Parallelise cross-format I/O

Concurrent disk reads accelerate the kind of work covered under cross-format parity testing. Use aiofiles for the read and GDAL’s /vsimem/ virtual filesystem to hand the bytes to OGR/GDAL without touching disk twice.

import asyncio
from pathlib import Path
import aiofiles
from osgeo import gdal

gdal.UseExceptions()

@pytest.mark.asyncio(loop_scope="session")
async def test_cross_format_parity():
    async def read_and_validate(path: str) -> bool:
        async with aiofiles.open(path, "rb") as f:
            data = await f.read()
        # Multi-file formats (Shapefile .shx/.dbf) need their sidecars staged in /vsimem/ too.
        vsi = f"/vsimem/{Path(path).name}"
        gdal.FileFromMemBuffer(vsi, data)
        try:
            return gdal.OpenEx(vsi) is not None
        finally:
            gdal.Unlink(vsi)

    results = await asyncio.gather(
        read_and_validate("test.shp"),
        read_and_validate("test.tif"),
    )
    assert all(results), "Cross-format parity check failed"

Matching fixture scope to loop scope

The second-largest source of instability is a fixture whose lifetime does not match the event loop’s. An asyncpg pool created in one loop and used from another raises an error that names neither the fixture nor the loop, and the message sends most people looking at the database.

Which fixture-scope and loop-scope combinations work A two-by-two comparison of pool scope against event loop scope. Function-scoped pool with function-scoped loop: valid, but connection setup is paid on every test. Session-scoped pool with function-scoped loop: invalid, because connections are bound to the loop that opened them and the second test meets a pool attached to a closed loop. Session-scoped pool with session-scoped loop: valid and recommended, since setup happens once and every test shares the same loop. Function-scoped pool with session-scoped loop: valid but wasteful. A closing rule states that a pool must never outlive the loop that created it. LOOP: FUNCTION-SCOPED LOOP: SESSION-SCOPED POOL: function POOL: session valid connection setup on every test correct but slow valid pool rebuilt inside one loop wasteful, not wrong invalid second test meets a pool bound to a closed loop valid — recommended setup paid once one loop for the whole session ← use this The rule behind the grid: a pool must never outlive the loop that created it — and the error when it does names neither the pool nor the loop.

The practical instruction is therefore to declare the loop scope explicitly rather than accepting the default, and to give every loop-bound resource the same scope as the loop it belongs to. That includes anything less obvious than a connection pool: an HTTP client session, a queue, a semaphore, and any object holding a future. Each of them captures the running loop at construction, and each produces the same unhelpful error when reused across a boundary.

Verify the fix

The fastest confirmation that the loop is never blocked is a probe test: schedule a heartbeat coroutine alongside an offloaded CPU task and assert the heartbeat kept ticking. If the geometry work were run inline, the heartbeat count would be zero.

import asyncio
import pytest

@pytest.mark.asyncio(loop_scope="session")
async def test_loop_is_not_blocked():
    ticks = 0
    async def heartbeat():
        nonlocal ticks
        for _ in range(5):
            await asyncio.sleep(0.01)
            ticks += 1

    def cpu_bound():            # stand-in for shapely.overlay / is_valid
        return sum(i * i for i in range(2_000_00))

    hb = asyncio.create_task(heartbeat())
    await asyncio.to_thread(cpu_bound)   # offloaded -> loop stays free
    await hb
    assert ticks == 5, "event loop was blocked during CPU-bound spatial work"

Run the suite as a pre-merge gate with pytest -q -m spatial_async; a non-zero exit blocks the merge. To surface hidden async defects in CI, export PYTHONASYNCIODEBUG=1 before the run — it logs unawaited coroutines and unclosed spatial file handles that otherwise pass silently.

Failure modes and edge cases

  1. Session pool on a per-function loop. The most common failure: a loop_scope="function" test consuming a loop_scope="session" pool raises got Future attached to a different loop. Align the test’s loop_scope with the fixture’s, or scope the pool to function and accept the reconnect cost.
  2. xdist workers sharing a spatial index. Under pytest-xdist, two workers writing the same R-tree or GDAL cache corrupt it non-deterministically. Use --dist=loadfile so a file’s tests stay on one worker, and give each worker its own cache directory via PYTEST_XDIST_WORKER.
  3. GDAL exceptions swallowed by gather. asyncio.gather without return_exceptions=False will propagate the first error, but a GDAL C-level abort can bypass Python’s exception machinery and kill the worker outright. Call gdal.UseExceptions() once at import so OGR errors become catchable RuntimeErrors.
  4. Empty result set, vacuous pass. A topology query returning zero rows makes gather(*[]) succeed instantly and the assertion trivially true. Assert len(rows) > 0 (or mark the layer as expected-empty) so an upstream query bug cannot disguise itself as a clean gate.
  5. CRS-unit mismatch before the async fan-out. Tolerance and ST_DWithin radii are in CRS units; mixing a degree-based EPSG:4326 geometry with a metre-based threshold yields false passes that no amount of concurrency will catch. Validate the CRS synchronously — and derive the radius the way you would when setting up spatial tolerance thresholds in assertions — before entering the async context.

By pinning asyncio_mode = "strict", scoping the asyncpg pool to its loop, offloading every CPU-bound Shapely/GDAL call through asyncio.to_thread, and proving the loop stays free with a heartbeat probe, concurrent spatial validation becomes a deterministic, high-throughput gate rather than a source of flaky CI — the same relationship-level contract that the broader catalogue of topology rule enforcement is built on.