aiofiles

aiofiles#

Test Data Workbench writes a deploy’s entire output, generator modules, YAML configs, and Markdown templates, to disk in one call. aiofiles is the library that does one narrow job in that path: give an async def function a way to write those files without calling the blocking builtin open() directly inside a coroutine.

Why aiofiles#

pyproject.toml pins aiofiles>=23.2.1 as a plain runtime dependency, listed alongside sqlalchemy and faker, not under [project.optional-dependencies]. It is imported in exactly one place in this codebase: src/test_data_workbench/adaptation/rapid_deployment.py, inside RapidAdapter._write_file_async.

That method exists because RapidAdapter.analyze_and_adapt is itself a coroutine, reachable both from a one-shot CLI command and from a FastAPI endpoint (see How it fits this project). Writing dozens of generated files with the builtin open() inside that coroutine would work; it would also block whatever else is sharing that coroutine’s event loop for as long as each write takes. aiofiles exists to remove that specific risk without rewriting the write path as callback-based or thread-managed code by hand.

The idea underneath#

It is worth being precise about what aiofiles actually changes, because the name invites the wrong mental model. Network sockets are asynchronous at the operating system level: Linux’s epoll, and the BSD/macOS kqueue that Python’s own event loop is built on, let a process register a set of file descriptors and be woken up only when one of them is ready, with no thread blocked in the meantime. Regular filesystem operations, open, read, write, on most operating systems, have no equivalent registration mechanism. There is no epoll event for “this disk write has finished.”

libuv, the C library behind Node.js’s event loop, states this plainly in its own design documentation, under “File I/O”: “Unlike network I/O, there are no platform-specific file I/O primitives libuv could rely on, so the current approach is to run blocking file I/O operations in a thread pool.” aiofiles’s own README makes the same point for Python specifically: “Ordinary local file IO is blocking, and cannot easily and portably be made asynchronous,” and describes its own approach as “delegating operations to a separate thread pool.”

That is the entire mechanism. aiofiles.open() does not make a disk faster or a write non-blocking at the kernel level. It hands the blocking open/read/write/close calls to a worker thread (by default, the event loop’s own executor, though aiofiles.open() accepts optional loop and executor arguments to override that), and returns an awaitable that resolves once that thread finishes. The coroutine that called it, and everything else scheduled on that same event loop, is free to keep running while the write happens on the worker thread rather than on the thread running the event loop.

Python’s own asyncio documentation describes exactly this pattern, independent of aiofiles, as the general answer to blocking code inside a coroutine. Its “Running Blocking Code” section warns that blocking code called directly “delays all concurrent asyncio Tasks and IO operations,” and recommends an executor, via loop.run_in_executor(), to run that code “in a different thread… to avoid blocking the OS thread with the event loop.” asyncio.to_thread(), added in Python 3.9, wraps the same idea in a smaller call: “Asynchronously run function func in a separate thread… Return a coroutine that can be awaited to get the eventual result.” aiofiles is, underneath its file-object-shaped API, a thin layer over this exact mechanism, specialized to file operations.

The practical consequence for a project like this one is not about disk throughput. It is that a plain synchronous file_path.write_text(...) call placed inside an async def function stalls that function’s entire event loop for the duration of the write, and every other coroutine waiting on that same loop, whether that is another concurrent HTTP request or another task in an asyncio.gather(), waits with it.

How it fits this project#

RapidAdapter.analyze_and_adapt (adaptation/rapid_deployment.py) is the single implementation behind three call sites:

  • tdw deploy, the CLI command in cli/deploy.py, which wraps a single call in asyncio.run(deploy(...)) and exits when it returns.

  • POST /v4/deployment/rapid, a FastAPI route in api/endpoints/deployment.py, mounted in api/main.py under the /v4/deployment prefix.

  • POST /v4/deployment/rapid/benchmark, the same endpoint module’s benchmark route, which calls analyze_and_adapt against a demo connection string.

All three paths converge on the same three-phase pipeline, and phase three ends by calling _write_output_files, which in turn calls _write_file_async once per generated file. That single method is where aiofiles is imported and used.

Writing generated output during deployment#

_write_output_files collects everything a deploy produces, per-table generator modules, configuration files, and team templates, into one dictionary keyed by relative path, then fires off one write per entry concurrently:

async def _write_output_files(self, output_path: Path,
                             generated_code: Dict[str, str],
                             config_files: Dict[str, str],
                             templates: Dict[str, str]) -> None:
    """Write all generated files to output directory."""
    all_files = {
        **{f'generators/{name}.py': code for name, code in generated_code.items()},
        **config_files,
        **templates
    }

    write_tasks = []
    for relative_path, content in all_files.items():
        full_path = output_path / relative_path
        write_tasks.append(self._write_file_async(full_path, content))

    await asyncio.gather(*write_tasks, return_exceptions=True)

generated_code holds one entry per table (generators/<table>.py), config_files holds config.yaml, .env.template, contribution_guide.yaml, and one scenarios/<name>.yaml per business scenario from ConfigurationBuilder, and templates holds README.md, CONTRIBUTING.md, three skill-level generator templates per entity type (generators/templates/<entity>_<level>.py, for user, customer, product, order, review), and three scenario templates. For a schema with several tables, that is easily twenty or more files, all handed to asyncio.gather at once rather than written one after another. _create_fallback_result, the emergency path taken when the main pipeline raises, calls the same _write_output_files with a smaller, fixed set of emergency generators and a single emergency_config.yaml.

The async context manager pattern#

_write_file_async is the only place aiofiles is imported, and the import itself is local to the function rather than a module-level import:

async def _write_file_async(self, file_path: Path, content: str) -> None:
    """Asynchronously write file with directory creation."""
    try:
        file_path.parent.mkdir(parents=True, exist_ok=True)

        # Use aiofiles for true async I/O if available, otherwise fallback
        try:
            import aiofiles
            async with aiofiles.open(file_path, 'w', encoding='utf-8') as f:
                await f.write(content)
        except ImportError:
            # Fallback to sync I/O
            with open(file_path, 'w', encoding='utf-8') as f:
                f.write(content)

    except Exception as e:
        self.logger.error(f"Failed to write {file_path}: {e}")

aiofiles.open() returns an async context manager whose shape mirrors the builtin’s line for line: async with ... as f in place of with ... as f, and await f.write(content) in place of f.write(content). The directory-creation step, file_path.parent.mkdir(parents=True, exist_ok=True), is plain synchronous pathlib, not wrapped in anything async; only the write itself goes through the thread pool. Both the success path and the whole outer try are defensive: an error partway through, a bad path, a permissions failure, is caught, logged through self.logger.error, and swallowed rather than propagated. That behavior is exercised directly in tests/test_rapid_deployment.py: test_write_file_async writes into a freshly created subdirectory and asserts the content round-trips, and test_write_file_async_bad_path writes to an unwritable path and asserts only that no exception escapes. Because _write_output_files gathers with return_exceptions=True on top of that, one failed file in a batch of twenty does not stop the other nineteen from being written.

A synchronous fallback that pyproject.toml makes unreachable#

The except ImportError branch above exists to fall back to the builtin open() if aiofiles is not installed. In this project, that branch cannot run in practice: aiofiles>=23.2.1 is a hard entry in pyproject.toml’s top-level dependencies list, the same list that holds fastapi, sqlalchemy, and pydantic. Anywhere test-data-workbench installs successfully at all, via pip install test-data-workbench or from this repository with its dependency closure resolved, aiofiles is already present, and import aiofiles cannot raise ImportError. The try/except here reads as defensive code written for a context, aiofiles as an optional accelerator, that this project’s own packaging does not actually create.

Sharp edges and limits#

The gather is where the real benefit lives, not the CLI invocation by itself. tdw deploy runs one asyncio.run(deploy(...)) call and exits; nothing else is competing for that event loop across the life of the process, so “aiofiles keeps the loop free for other work” is not, on its own, a reason this CLI path needs it. What the CLI path does get is narrower: _write_output_files hands asyncio.gather a batch that can run to twenty or more files, and dispatching each one to a worker thread lets their open/write/close syscalls overlap instead of running strictly one after another. That is a real, if modest, win, and it would exist with a plain concurrent.futures.ThreadPoolExecutor.map() just as well; nothing about it requires the file to be part of an asyncio application.

The API path is the one where blocking would actually matter, and it is labeled a development server. The same _write_file_async also backs POST /v4/deployment/rapid. api/main.py’s __main__ block, under the comment “Development server runner,” calls uvicorn.run(..., reload=args.reload) with no workers argument, Uvicorn’s single-worker default. Under one worker, a blocking write inside one deployment’s request handler would stall every other request that server is holding at the same time, including GET /v4/deployment/rapid/status for a different deployment. That is the scenario aiofiles is actually built to prevent. Nothing in this repository configures a production ASGI deployment (no multi-worker Uvicorn or Gunicorn configuration, no Dockerfile, no process manager entry) that would exercise more than one worker, so this benefit is real in principle but currently unproven under this project’s own deployment setup.

Everywhere else in this codebase, writing generated text to disk is synchronous, and that is the correct choice there, not an oversight. cli/team.py writes a workspace README.md, a pre-commit hook, a config_file, quick_start.md, and assignments.md, all with plain open(path, "w"). core/demo_coordinator.py’s export_demo_plan writes a JSON report the same way. cli/analyze.py writes both a JSON results file and a YAML analysis-summary file with plain open(path, "w"). None of those five call sites live inside an async def function, so there is no event loop for a blocking call to stall in the first place; aiofiles would add a dependency and a thread hop there for no benefit. The result is a project where “generated output gets written asynchronously” is true of exactly one function and false everywhere else, which is defensible given the different context each call site runs in, but it is not a rule the code states anywhere, and a reader skimming for “how does this project write files” needs to know the answer is “it depends which module.”

The standard library has covered this since before the project’s own minimum Python version. pyproject.toml requires python_requires=">=3.10". asyncio.to_thread() shipped in Python 3.9, one full minor version earlier. await asyncio.to_thread(file_path.write_text, content, encoding='utf-8') delegates to the same executor-backed thread pool as aiofiles.open(), with no extra dependency. What aiofiles buys over that stdlib option is not new capability; it is a file-object-shaped API, async with aiofiles.open(path, 'w') as f: await f.write(content) reads like the familiar synchronous with open(...) block, where a to_thread call reads like a function dispatch wrapped around one. That is a genuine readability argument, particularly next to the plain with open(...) blocks in cli/team.py and cli/analyze.py, but it is an ergonomic tradeoff, not a functional gap in the standard library.

Learn more resources#

Official documentation#

  • Tinche/aiofiles README: the project’s own explanation of why file IO is blocking, how it delegates to a thread pool, the async context manager usage shown above, and the optional loop/executor arguments to aiofiles.open().

  • aiofiles on PyPI: current release metadata, useful for checking the installed version against this project’s aiofiles>=23.2.1 pin.

  • asyncio.to_thread: the standard-library equivalent mechanism, available since Python 3.9.

  • loop.run_in_executor: the lower-level executor API that both to_thread and aiofiles build on.

  • Developing with asyncio: Running Blocking Code: Python’s own documentation on why blocking calls inside a coroutine delay every other concurrent task on the same event loop.

Tutorials and blogs#

Videos#