io-adapters documentation¶

Motivation¶

Testing use cases that involve I/O is inherently difficult because they depend on:

external state (filesystems, databases, services)
side effects that are hard to observe directly
slow or flaky infrastructure

A common mitigation is to combine:

Dependency Injection (DI)
The Repository / Adapter pattern

This allows business logic to depend on an abstract interface rather than concrete I/O.

However, in practice this usually requires:

writing and maintaining bespoke fake implementations
keeping fake behaviour in sync with real implementations
duplicating boilerplate across domains

For small or medium-sized projects, this overhead can outweigh the benefits.

Simply register each I/O function with one of the register decorators and the functionality will be added to the RealAdapter object, on top of that a stub will be added to the FakeAdapter object too so you can pass in either to your usecase and the functionality will work.

Example¶

from enum import Enum
from pathlib import Path

from io_adapters import (
    IoAdapter,
    RealAdapter,
    add_domain,
    get_fake_adapter,
    get_real_adapter,
    register_domain_read_fn,
    register_domain_write_fn,
)


# you can use any hashable object to register an I/O function
class FileFormat(Enum):
    JSON = "json"


add_domain("orders")
add_domain("payment")


@register_domain_read_fn("orders", "str")
def read_str(path: str | Path, **kwargs: dict) -> str: ...


# stack decorators to register the same function to multiple domains
@register_domain_read_fn("orders", FileFormat.JSON)
@register_domain_read_fn("payment", FileFormat.JSON)
def read_json(path: str | Path, **kwargs: dict) -> dict: ...


@register_domain_write_fn("orders", "str")
def write_str(data: dict, path: str | Path, **kwargs: dict) -> None: ...


@register_domain_write_fn("orders", FileFormat.JSON)
@register_domain_write_fn("payment", FileFormat.JSON)
def write_json(data: dict, path: str | Path, **kwargs: dict) -> None: ...


def some_usecase(adapter: IoAdapter, path: str) -> None:
    adapter.read(path, "str")
    # Some business logic
    new_path = f"{path}_new.json"

    adapter.write({"a": 1}, new_path, FileFormat.JSON)


# in production inject the real adapter
orders_adapter: RealAdapter = get_real_adapter("orders")
some_usecase(orders_adapter, "some/path/to/file.json")


# in testing inject the fake which has all the same funcitonality as the
# `RealAdapter` and assert that the fakes end state is as expected
fake_adapter = get_fake_adapter("orders")
some_usecase(fake_adapter, "some/path/to/file.json")
assert fake_adapter.files["some/path/to/file.json"] == {"a": 1}

class io_adapters.Container(domains: Iterable)¶

Bases: object

Registry and factory for domain-scoped I/O adapters.

The Container provides a central registry for I/O functions that are grouped by domain. Each domain has isolated access to a specific set of read and write functions, allowing different parts of an application to share common I/O implementations without violating domain boundaries.

Domain isolation¶

In some cases, different domains need isolated access to the same underlying functionality.

For example:

The orders domain may:
- read json files
- write parquet files
The reporting domain may:
- read from a database
- write parquet files

While both domains can reuse a shared write_parquet implementation, they must not have access to each other’s read capabilities. The Container enforces this isolation by maintaining separate registries per domain.

Design¶

The Container:

Maintains a mapping of domains to read/write function registries
Provides decorator-style registration APIs for read and write functions
Acts as a factory for creating domain-specific adapters

Two adapter types are supported:

RealAdapter for production usage
FakeAdapter for testing, allowing stateful simulation of I/O

Testing¶

Using a FakeAdapter makes it possible to simulate external I/O while keeping all state in memory. This allows tests to:

Accumulate state across reads and writes
Assert against the final external state
Avoid filesystem or network dependencies entirely

This results in tests that are faster, deterministic, and easier to reason about.

Usage overview¶

Define a Container with one or more domains
Register read/write functions per domain
Request either a real or fake adapter for a given domain
Inject the adapter into application code

The Container itself is intentionally simple and does not perform any: I/O; it only wires together domain-specific capabilities.

Tip

Usage of a custom Container is only recommended if you have a complex set of domains and need multiple ``Container``s initialised at the same time.

If you have a simple set of domains you can use the convenience functions with the same names as the Container methods which will be added to the default Container which is an instance you don’t need to initialise yourself.

add_domain(domain: Hashable) → None¶

Add a domain to a Container

from io_adapters import Container

container = Container()
container.add_domain("orders")

The orders domain is now added to the Container and can have IO functions registered to it.

Relying on deliberate registering of a domain avoids situations where a typo could register a function to a non-existent domain: e.g. 'ordesr' instead of the intended 'orders'.

Domains can also be passed into the Container on initialisation.

container = Container(domains=["orders"])

domain_fns: dict[Hashable, dict[Hashable, Callable]]¶

domains: Iterable¶

get_fake_adapter(domain: Hashable, files: dict | None = None) → FakeAdapter¶

Get a FakeAdapter for the given domain.

The returned adapter will have all of the functions registered to that domain.

from io_adapters import FakeAdapter, Container

container = Container(domains=["orders"])
orders_adapter: FakeAdapter = container.get_fake_adapter("orders")

The FakeAdapter that is assigned to the orders_adapter variable will have fake representations for all of the registered read and write I/O functions.

This can optionally be given a dictionary of files to setup the initial state for testing. An example of how this could be used is below:

from io_adapters import FakeAdapter, Container

starting_files = {"path/to/data.json": {"a": 0, "b": 1}}

container = Container(domains=["orders"])
orders_adapter: FakeAdapter = container.get_fake_adapter("orders", starting_files)

some_orders_usecase(adapter=orders_adapter, data_path="path/to/data.json")

assert orders_adapter.files["path/to/modified_data.json"] == {"a": 1, "b": 2}

get_real_adapter(domain: Hashable) → RealAdapter¶

Get a RealAdapter for the given domain from a Container.

The returned adapter will have all of the functions registered to that domain.

from io_adapters import RealAdapter, Container

container = Container(domains=["orders"])
orders_adapter: RealAdapter = container.get_real_adapter("orders")

The RealAdapter that is assigned to the orders_adapter variable will have all of the registered read and write I/O functions.

register_domain_read_fn(domain: Hashable, key: Hashable) → Callable¶

Register a read function to a domain in a Container.

Decorators can be stacked to register the same function to multiple domains.

from io_adapters import Container

container = Container(domains=["orders", "payment"])

@container.register_domain_read_fn("orders", "str")
def read_str(path: str | Path, **kwargs: dict) -> str:
    ...


@container.register_domain_read_fn("orders", "json")
@container.register_domain_read_fn("payment", "json")
def read_json(path: str | Path, **kwargs: dict) -> dict:
    ...

register_domain_write_fn(domain: Hashable, key: Hashable) → Callable¶

Register a write function to a domain in a Container.

Decorators can be stacked to register the same function to multiple domains.

from io_adapters import Container

container = Container(domains=["orders"])

@container.register_domain_write_fn("orders", "str")
def write_str(data: dict, path: str | Path, **kwargs: dict) -> None:
    ...

container.add_domain("payment")

@container.register_domain_write_fn("orders", "json")
@container.register_domain_write_fn("payment", "json")
def write_json(data: dict, path: str | Path, **kwargs: dict) -> None:
    ...

class io_adapters.FakeAdapter(read_fns={'json': <function read_json>}, write_fns={'json': <function write_json>}, guid_fn: Callable[[], str]=None, datetime_fn: Callable[[], ~datetime.datetime]=None, files: dict[str, Data]=NOTHING)¶

Bases: IoAdapter

files: dict[str, Data]¶

get_datetime() → datetime¶

get_guid() → str¶

class io_adapters.IoAdapter(read_fns={'json': <function read_json>}, write_fns={'json': <function write_json>}, guid_fn: Callable[[], str]=None, datetime_fn: Callable[[], ~datetime.datetime]=None)¶

Bases: object

datetime_fn: Callable[[], datetime]¶

get_datetime() → datetime¶

get_guid() → str¶

guid_fn: Callable[[], str]¶

read(path: str | Path, file_type: Hashable, **kwargs: dict) → Data¶

Read path using the registered function for file_type.

Raises:: NotImplementedError – If the given file_type does not have a registered function.

Usage¶

Here the read_json function is registered with the register_read_fn decorator.

Then when the adapter object calls read with the "json" file_type it will use the registered function.

The key used to register the function doesn’t have to be a string, as long as it’s Hashable it can be used.

from io_adapters import RealAdapter, register_read_fn

@register_read_fn("json")
def read_json(path: str | Path, **kwargs: dict) -> dict:
    return json.loads(Path(path).read_text(), **kwargs)

adapter = RealAdapter()
data = adapter.read("some/path/to/file.json", "json")

read_fns: MappingProxyType¶

write(data: Data, path: str | Path, file_type: Hashable, **kwargs: dict) → None¶

Write data to path using the registered function for file_type.

Raises:: NotImplementedError – If the given file_type does not have a registered function.

Usage¶

Here the write_json function is registered with the register_write_fn decorator.

Then when the adapter object calls write with the WriteFormat.JSON file_type it will use the registered function.

from enum import Enum
from io_adapters import RealAdapter, register_write_fn

class WriteFormat(Enum):
    JSON = "json"

@register_write_fn(WriteFormat.JSON)
def write_json(data: dict, path: str | Path, **kwargs: dict) -> None:
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(data, **kwargs))

adapter = RealAdapter()
adapter.write({"a": 1}, "some/path/to/file.json", WriteFormat.JSON)

fake_adapter = FakeAdapter()
fake_adapter.write({"a": 1}, "some/path/to/file.json", WriteFormat.JSON)

The interfaces between the FakeAdapter and the RealAdapter means that the two can be passed in interchangeably, making testing much easier

def some_usecase(adapter: IoAdapter, path: str) -> None:
    # Some business logic

    adapter.write({"a": 1}, path, WriteFormat.JSON)

# in production inject the real adapter
some_usecase(RealAdapter(), "some/path/to/file.json")

# in testing inject the fake and assert that the fakes end state is as expected
fake = FakeAdapter()
some_usecase(fake, "some/path/to/file.json")
assert fake.files["some/path/to/file.json"] == {"a": 1}

write_fns: MappingProxyType¶

class io_adapters.RealAdapter(read_fns={'json': <function read_json>}, write_fns={'json': <function write_json>}, guid_fn: Callable[[], str]=None, datetime_fn: Callable[[], ~datetime.datetime]=None)¶: Bases: IoAdapter

io_adapters.add_domain(domain: Hashable) → None¶

Add a domain to the default Container

from io_adapters import add_domain

add_domain("orders")

The orders domain is now added to the default Container and can have IO functions registered to it.

Relying on deliberate registering of a domain avoids situations where a typo could register a function to a non-existent domain: e.g. 'ordesr' instead of the intended 'orders'.

io_adapters.get_fake_adapter(domain: Hashable, files: dict | None = None) → FakeAdapter¶

Get a FakeAdapter for the given domain.

The returned adapter will have all of the functions registered to that domain.

from io_adapters import FakeAdapter, get_fake_adapter

orders_adapter: FakeAdapter = get_fake_adapter("orders")

The FakeAdapter that is assigned to the orders_adapter variable will have fake representations for all of the registered read and write I/O functions.

This can optionally be given a dictionary of files to setup the initial state for testing. An example of how this could be used is below:

from io_adapters import FakeAdapter, get_fake_adapter

starting_files = {"path/to/data.json": {"a": 0, "b": 1}}

orders_adapter: FakeAdapter = get_fake_adapter("orders", starting_files)

some_orders_usecase(adapter=orders_adapter, data_path="path/to/data.json")

assert orders_adapter.files["path/to/modified_data.json"] == {"a": 1, "b": 2}

io_adapters.get_real_adapter(domain: Hashable) → RealAdapter¶

Get a RealAdapter for the given domain.

The returned adapter will have all of the functions registered to that domain.

from io_adapters import RealAdapter, get_real_adapter

orders_adapter: RealAdapter = get_real_adapter("orders")

The RealAdapter that is assigned to the orders_adapter variable will have all of the registered read and write I/O functions.

io_adapters.register_domain_read_fn(domain: Hashable, key: Hashable) → Callable¶

Register a read function to a domain in the default Container.

Decorators can be stacked to register the same function to multiple domains.

from io_adapters import add_domain, register_domain_read_fn

add_domain("orders")

@register_domain_read_fn("orders", "str")
def read_str(path: str | Path, **kwargs: dict) -> str:
    ...

add_domain("payment")

@register_domain_read_fn("orders", "json")
@register_domain_read_fn("payment", "json")
def read_json(path: str | Path, **kwargs: dict) -> dict:
    ...

io_adapters.register_domain_write_fn(domain: Hashable, key: Hashable) → Callable¶

Register a write function to a domain in the default Container.

Decorators can be stacked to register the same function to multiple domains.

from io_adapters import add_domain, register_domain_write_fn

add_domain("orders")

@register_domain_write_fn("orders", "str")
def write_str(data: dict, path: str | Path, **kwargs: dict) -> None:
    ...

add_domain("payment")

@register_domain_write_fn("orders", "json")
@register_domain_write_fn("payment", "json")
def write_json(data: dict, path: str | Path, **kwargs: dict) -> None:
    ...

io_adapters.register_read_fn(key: Hashable) → Callable¶

Register a read function to the read functions constant.

This is useful for smaller projects where domain isolation isn’t required.

from io_adapters import RealAdapter, register_read_fn

@register_read_fn("json")
def read_json(path: str | Path, **kwargs: dict) -> dict:
    ...

This function will be accessible when you initialise a RealAdapter: and a stub of the functionality will be added to a FakeAdapter.

io_adapters.register_write_fn(key: Hashable) → Callable¶

Register a write function to the write functions constant.

This is useful for smaller projects where domain isolation isn’t required.

from enum import Enum
from io_adapters import RealAdapter, register_write_fn

class WriteFormat(Enum):
    JSON = "json"

@register_write_fn(WriteFormat.JSON)
def write_json(data: dict, path: str | Path, **kwargs: dict) -> None:
    ...

This function will be accessible when you initialise a RealAdapter: and a stub of the functionality will be added to a FakeAdapter.

io-adapters documentation¶

Motivation¶

Example¶

Domain isolation¶

Design¶

Testing¶

Usage overview¶

Usage¶

Usage¶

Indices and tables¶

Table of Contents

This Page