Running async spatial tests with pytest-asyncio

7 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.

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"

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.