anyio 4.14.2: from_thread.BlockingPortal, from_thread.run

Clean-room public code sample — generation instructions

Write a brand-new, minimal, self-contained code sample in this clean-room directory.
Do not copy, paraphrase, or reference any existing project source. Work only from this spec.

A csx.json manifest scaffold already exists. Do not recreate it from memory. Preserve its case.goal, packages and symbols; fill its empty case.contract with exact assertions and correct its environment, commands and verifierAdapter for the files you generate.

Goal: verify pkg:pypi/anyio@4.14.2
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/anyio@4.14.2
Demonstrate these symbols/APIs:
  - anyio.from_thread.BlockingPortal
  - anyio.from_thread.run

Rules:
  - One focused purpose; the smallest project that proves the goal.
  - Include a contract test (test/contract.*) that runs OFFLINE and exits 0 exactly when the goal behavior works.
  - Pin every dependency with a lockfile so resolution is reproducible.
  - No secrets, credentials, or tokens. No real URLs (only example.com or localhost). No absolute paths.
  - No personal names, emails, company names, or project identifiers of any kind.
  - No binaries and no generated output (node_modules, dist, target, venv, .git, .env).
  - Keep it under 200 files and 256KB packed.

{"case":{"caseId":"case:sha256:7fb1d037c5e252c167d412047fbe1cb641faa247fbed049bb63820dabac1035b","contract":["anyio.from_thread.BlockingPortal allows synchronous external threads to call async coroutines in the portal event loop and synchronously obtain results","anyio.from_thread.BlockingPortal.start_task_soon schedules concurrent background tasks in the portal event loop without blocking synchronous callers","anyio.from_thread.run executes an async coroutine in the event loop thread from a worker thread spawned via anyio.to_thread.run_sync and returns its result","anyio.from_thread.run_sync executes a synchronous callable in the event loop thread from a worker thread"],"goal":"verify pkg:pypi/anyio@4.14.2","kind":"HOW","packages":["pkg:pypi/anyio@4.14.2"],"schemaVersion":1,"symbols":["anyio.from_thread.BlockingPortal","anyio.from_thread.run"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"pypi","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"pip","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/anyio@4.14.2"],"schemaVersion":1,"subject":"pkg:pypi/anyio@4.14.2","symbols":["anyio.from_thread.BlockingPortal","anyio.from_thread.run"],"verifierAdapter":"python@1"}

anyio==4.14.2
idna==3.10
sniffio==1.3.1
typing-extensions==4.12.2

"""AnyIO from_thread portal and execution utilities."""

from typing import Any, Callable, TypeVar
import anyio
import anyio.from_thread

T = TypeVar("T")


async def async_double(value: int) -> int:
    """Async helper that multiplies a value by 2."""
    await anyio.sleep(0.001)
    return value * 2


async def async_fetch_data(key: str) -> dict[str, Any]:
    """Async helper returning a structured dictionary."""
    await anyio.sleep(0.001)
    return {"key": key, "status": "active"}


def sync_portal_runner(portal: anyio.from_thread.BlockingPortal, key: str) -> dict[str, Any]:
    """Run an async coroutine from a synchronous function using a BlockingPortal."""
    return portal.call(async_fetch_data, key)


async def run_worker_with_event_loop_callbacks() -> tuple[int, int]:
    """Demonstrate worker thread calling back into the event loop thread via from_thread."""
    def worker() -> tuple[int, int]:
        # Call an async function back on the event loop thread
        async_res = anyio.from_thread.run(async_double, 21)
        # Call a sync function back on the event loop thread
        sync_res = anyio.from_thread.run_sync(lambda x: x + 10, 50)
        return async_res, sync_res

    return await anyio.to_thread.run_sync(worker)


if __name__ == "__main__":
    with anyio.from_thread.start_blocking_portal() as p:
        res = p.call(async_double, 10)
        print(f"Portal call result: {res}")

{
  "schemaVersion": 1,
  "goal": "verify pkg:pypi/anyio@4.14.2",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/anyio@4.14.2"
  ],
  "symbols": [
    "anyio.from_thread.BlockingPortal",
    "anyio.from_thread.run"
  ]
}

"""Contract verification test for anyio.from_thread."""

import os
import sys
import threading

# Ensure project root is on sys.path for local module imports
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import anyio
import anyio.from_thread
from sample import async_double, async_fetch_data, sync_portal_runner, run_worker_with_event_loop_callbacks


def test_blocking_portal_call() -> None:
    """Assertion 1: anyio.from_thread.BlockingPortal allows synchronous external threads to call async coroutines in the portal event loop and synchronously obtain results."""
    with anyio.from_thread.start_blocking_portal() as portal:
        # Call from the synchronous main thread
        result = sync_portal_runner(portal, "item-42")
        assert result == {"key": "item-42", "status": "active"}, f"Expected dict matching item-42, got {result}"

        # Call with arguments
        num = portal.call(async_double, 25)
        assert num == 50, f"Expected 50, got {num}"

        # Error propagation
        async def failing_coro() -> None:
            raise ValueError("portal error test")

        caught = False
        try:
            portal.call(failing_coro)
        except ValueError as exc:
            caught = True
            assert str(exc) == "portal error test"

        assert caught is True, "Expected ValueError to propagate to synchronous caller"


def test_blocking_portal_start_task_soon() -> None:
    """Assertion 2: anyio.from_thread.BlockingPortal.start_task_soon schedules concurrent background tasks in the portal event loop without blocking synchronous callers."""
    with anyio.from_thread.start_blocking_portal() as portal:
        task_executed = False
        event = threading.Event()

        async def background_task(flag_value: bool) -> None:
            nonlocal task_executed
            await anyio.sleep(0.01)
            task_executed = flag_value
            event.set()

        portal.start_task_soon(background_task, True)
        # Task runs asynchronously in portal loop; synchronous caller is not blocked
        assert event.wait(timeout=2.0) is True, "Background task should complete within timeout"
        assert task_executed is True, "Expected task_executed to be True"


def test_from_thread_run_and_run_sync() -> None:
    """Assertion 3: anyio.from_thread.run executes an async coroutine in the event loop thread from a worker thread spawned via anyio.to_thread.run_sync and returns its result."""
    """Assertion 4: anyio.from_thread.run_sync executes a synchronous callable in the event loop thread from a worker thread."""
    async def _test() -> None:
        async_res, sync_res = await run_worker_with_event_loop_callbacks()
        assert async_res == 42, f"Expected 42 from from_thread.run, got {async_res}"
        assert sync_res == 60, f"Expected 60 from from_thread.run_sync, got {sync_res}"

    anyio.run(_test)


def main() -> None:
    test_blocking_portal_call()
    test_blocking_portal_start_task_soon()
    test_from_thread_run_and_run_sync()
    print("Contract verification passed successfully.")


if __name__ == "__main__":
    main()