pyiv

Guice-style dependency injection for Python.

pyiv provides type-based constructor injection, scopes, qualified keys, and built-in test doubles. Runtime has zero third-party dependencies. Python 3.8+.

Key Features:

  • Type-based constructor injection from annotations

  • Scopes (per-injector and process-wide singletons, plus custom Scope)

  • Qualified keys and a fluent Binder API

  • Module install, private modules, child injectors, and config override

  • Map/Set/List multibinders; Stage.PRODUCTION eager singletons

  • Reflection to discover implementations in a package

  • Test doubles for Clock, Filesystem, Console, and DateTimeService

  • Zero runtime dependencies

Quick Start:

>>> from pyiv import Config, get_injector
>>> class Database:
...     pass
>>> class PostgreSQL(Database):
...     pass
>>> class MyConfig(Config):
...     def configure(self):
...         self.register(Database, PostgreSQL)
>>> injector = get_injector(MyConfig)
>>> isinstance(injector.inject(Database), PostgreSQL)
True
class pyiv.Base64SerDe[source]

Bases: SerDe

Base64 encoding SerDe.

Use this when binary payloads must travel as text (headers, logs, or text-only transports). Strings are UTF-8 encoded first; non-bytes objects fall back to pickle before encoding. Deserialize returns raw bytes.

Example

>>> from pyiv.serde.encodings import Base64SerDe
>>> serde = Base64SerDe()
>>> serde.handler_type
'base64'
>>> encoded = serde.serialize(b"hello")
>>> encoded
'aGVsbG8='
>>> serde.deserialize(encoded)
b'hello'
deserialize(data: str | bytes, target_type: Type[T] | None = None) → T[source]

Deserialize base64-encoded data.

Parameters:
  • data – The base64-encoded string or bytes

  • target_type – Optional type hint (returns bytes by default)

Returns:

Decoded bytes

property handler_type: str

Return the handler type identifier.

Returns:

The handler type identifier (“base64”)

serialize(obj: Any) → str[source]

Serialize using base64 encoding.

Parameters:

obj – The data to encode (bytes or string)

Returns:

Base64-encoded string

class pyiv.BaseConsole[source]

Bases: ABC

Abstract base class for console implementations.

Subclass this when you need a custom Console (file-like write plus optional TTY helpers). Implement write/flush/writable; terminal methods default to no-ops so simple sinks stay small. Prefer MemoryConsole for capturing print() in tests.

Why this exists: Shared base for Console implementations so TTY helpers default to no-ops.

Example

>>> from pyiv.console import BaseConsole
>>> class ListConsole(BaseConsole):
...     def __init__(self):
...         self.chunks = []
...     def write(self, s: str) -> int:
...         self.chunks.append(s)
...         return len(s)
...     def flush(self) -> None:
...         pass
...     def writable(self) -> bool:
...         return True
>>> c = ListConsole()
>>> print("hi", file=c)
>>> "".join(c.chunks)
'hi\n'
bold(enabled: bool = True) → None[source]

Enable or disable bold text (no-op by default).

clear() → None[source]

Clear the entire screen (no-op by default).

clear_line() → None[source]

Clear the current line (no-op by default).

clear_state() → None[source]

Clear all state (default: no-op).

clear_to_end_of_line() → None[source]

Clear from cursor to end of line (no-op by default).

abstract flush() → None[source]

Flush any buffered output.

get_color() → Tuple[int | None, int | None][source]

Get current color state (default: None, None).

get_cursor() → Tuple[int, int][source]

Get current cursor position (default: 0, 0).

get_cursor_position() → Tuple[int, int] | None[source]

Get current cursor position (default: None).

get_echo_enabled() → bool[source]

Get echo state (default: True).

get_events() → List[TerminalEvent][source]

Get event history (default: empty list).

get_raw_mode() → bool[source]

Get raw mode state (default: False).

get_screen() → List[List[str]][source]

Get screen buffer (default: empty).

get_screen_char(x: int, y: int) → str[source]

Get character at position from screen buffer (default: space).

Parameters:
  • x – Column (0-based)

  • y – Row (0-based)

Returns:

Character at position

get_screen_line(y: int) → str[source]

Get specific line from screen buffer (default: empty string).

Parameters:

y – Line number (0-based)

Returns:

Line contents as string

get_size() → Tuple[int, int][source]

Get terminal size (default: 80x24).

get_style() → Dict[str, bool][source]

Get text style state (default: all False).

Returns:

Dictionary of style flags

hide_cursor() → None[source]

Hide the cursor (no-op by default).

is_tty() → bool[source]

Check if output is a real terminal (default: False).

move_cursor(x: int, y: int) → None[source]

Move cursor to position (no-op by default).

move_cursor_down(n: int = 1) → None[source]

Move cursor down n lines (no-op by default).

move_cursor_home() → None[source]

Move cursor to home position (no-op by default).

move_cursor_left(n: int = 1) → None[source]

Move cursor left n columns (no-op by default).

move_cursor_right(n: int = 1) → None[source]

Move cursor right n columns (no-op by default).

move_cursor_up(n: int = 1) → None[source]

Move cursor up n lines (no-op by default).

read_char(timeout: float | None = None) → str | None[source]

Read a single character (default: None).

read_line(prompt: str = '') → str[source]

Read a line of input (default: empty string).

read_password(prompt: str = '') → str[source]

Read password with echo disabled (default: empty string).

reset_color() → None[source]

Reset colors to default (no-op by default).

restore_cursor() → None[source]

Restore saved cursor position (no-op by default).

save_cursor() → None[source]

Save current cursor position (no-op by default).

set_color(fg: int | None = None, bg: int | None = None) → None[source]

Set foreground and/or background color (no-op by default).

set_echo(enabled: bool) → None[source]

Turn echo on/off (no-op by default).

set_raw_mode(enabled: bool) → None[source]

Enable/disable raw mode (no-op by default).

show_cursor() → None[source]

Show the cursor (no-op by default).

underline(enabled: bool = True) → None[source]

Enable or disable underline (no-op by default).

abstract writable() → bool[source]

Check if console is writable.

Returns:

True if console supports writing

abstract write(s: str) → int[source]

Write string to console.

Parameters:

s – String to write

Returns:

Number of characters written

class pyiv.BaseFactory[source]

Bases: ABC, Generic[T]

ABC when you want a class-based factory with constructor injection.

Why this exists: Protocols are fine for typing; subclassing BaseFactory gives a real type the injector can construct, with dependencies supplied to __init__ and runtime args to create.

Example

>>> from pyiv.factory import BaseFactory
>>> class User:
...     def __init__(self, name: str):
...         self.name = name
>>> class UserFactory(BaseFactory[User]):
...     def create(self, name: str) -> User:
...         return User(name=name)
>>> UserFactory().create("Ada").name
'Ada'
abstract create(*args: Any, **kwargs: Any) → T[source]

Create an instance of type T.

Parameters:
  • *args – Positional arguments for instance creation

  • **kwargs – Keyword arguments for instance creation

Returns:

An instance of type T

class pyiv.BaseProvider[source]

Bases: ABC, Generic[T]

Abstract base class for provider implementations.

Why this exists: Subclassable provider when you need a real class the injector can construct.

Provides a concrete base class for providers that need to be instantiated. Subclasses should implement the get() method.

Example

>>> from pyiv.provider import BaseProvider, InstanceProvider
>>>
>>> class User:
...     def __init__(self, name: str):
...         self.name = name
>>>
>>> class UserProvider(BaseProvider[User]):
...     def get(self) -> User:
...         return User("alice")
>>>
>>> UserProvider().get().name
'alice'
>>> InstanceProvider(User("bob")).get().name
'bob'
abstract get() → T[source]

Get an instance of type T.

Returns:

An instance of type T

class pyiv.Binder(*args, **kwargs)[source]

Bases: Protocol

Fluent configuration API for contributing bindings to a Config.

Why this exists: Direct Config.register* calls work, but complex graphs read better as bind(X).to(Y).in_scope(Z). The binder also supports install, expose, and require_explicit_bindings.

Example

>>> from pyiv import Config, get_injector
>>> class Database:
...     pass
>>> class PostgreSQL(Database):
...     pass
>>> class MyConfig(Config):
...     def configure(self):
...         self.get_binder().bind(Database).to(PostgreSQL)
>>> isinstance(get_injector(MyConfig).inject(Database), PostgreSQL)
True
__init__(*args, **kwargs)
bind(abstract: Type[T]) → BindingBuilder[T][source]

Start a binding configuration.

Parameters:

abstract – The abstract type to bind

Returns:

A binding builder for fluent configuration

bind_instance(abstract: Type[T], instance: T) → None[source]

Bind to a pre-created instance.

Parameters:
  • abstract – The abstract type

  • instance – The pre-created instance

bind_key(key: Key[T]) → BindingBuilder[T][source]

Start a binding configuration with a qualified key.

Parameters:

key – The qualified key to bind

Returns:

A binding builder for fluent configuration

expose(type_or_key: Any) → None[source]

Expose a private binding to the parent environment.

Only meaningful on a PrivateConfig.

install(config: Any) → None[source]

Install another configuration module.

Parameters:

config – Another Config instance or subclass to install. Later installs overwrite the same type/key (last wins).

require_explicit_bindings() → None[source]

Require every injectable type to be bound explicitly (no JIT).

class pyiv.BindingBuilder(*args, **kwargs)[source]

Bases: Protocol, Generic[T]

Fluent steps after Binder.bind() (.to / .in_scope / …).

Why this exists: A single register(...) call buries scope and provider choices in kwargs. The builder makes the binding shape readable as a chain while still writing the same Config store.

Example

>>> from pyiv import Config, get_injector
>>> from pyiv.scope import SingletonScope
>>> class Database:
...     pass
>>> class PostgreSQL(Database):
...     pass
>>> class MyConfig(Config):
...     def configure(self):
...         self.get_binder().bind(Database).to(PostgreSQL).in_scope(SingletonScope())
>>> isinstance(get_injector(MyConfig).inject(Database), PostgreSQL)
True
__init__(*args, **kwargs)
in_scope(scope: Scope) → BindingBuilder[T][source]

Set the scope for this binding.

Parameters:

scope – The scope to use

Returns:

Self for method chaining

to(implementation: Type[T]) → BindingBuilder[T][source]

Bind to a concrete implementation.

Parameters:

implementation – The concrete class to bind to

Returns:

Self for method chaining

to_instance(instance: T) → BindingBuilder[T][source]

Bind to a pre-created instance.

Parameters:

instance – The instance to bind to

Returns:

Self for method chaining

to_provider(provider: Provider[T]) → BindingBuilder[T][source]

Bind to a provider.

Parameters:

provider – The provider to use for instance creation

Returns:

Self for method chaining

class pyiv.CLICommand(args: Namespace, injector: Any | None = None)[source]

Bases: Command

Base class for one-shot CLI commands (not long-running services).

Use this for tools that init, run once, and exit. Override run() (and optionally init()/cleanup()); set self._exit_code for non-zero status. Do not override execute() — it owns KeyboardInterrupt / SystemExit handling.

Example

>>> import argparse
>>> from pyiv.command import CLICommand
>>> class EchoCommand(CLICommand):
...     @classmethod
...     def get_name(cls) -> str:
...         return "echo"
...     def run(self) -> None:
...         self._exit_code = 0
>>> EchoCommand(argparse.Namespace()).execute()
0
__init__(args: Namespace, injector: Any | None = None)[source]

Initialize CLI command with parsed arguments.

cleanup() → None[source]

Cleanup after CLI command (no-op by default, override if needed).

execute() → int[source]

Execute the CLI command lifecycle.

Returns:

Exit code (0 for success, non-zero for error)

init() → None[source]

Initialize CLI command (no-op by default, override if needed).

run() → None[source]

Run the CLI command (no-op by default).

Override this method to implement command logic. Set self._exit_code to return a non-zero status from execute().

class pyiv.ChainHandler[source]

Bases: ABC

Pluggable handler looked up by chain type and handler name/type.

Why this exists: Apps need interchangeable strategies (JSON vs pickle, HTTP vs HTTPS) without hard-coding classes at call sites. Register handlers on Config and resolve them with inject_chain_handler.

Example

>>> from pyiv.serde import JSONSerDe
>>> h: ChainHandler = JSONSerDe()
>>> h.chain_type is ChainType.ENCODING
True
>>> h.handler_type
'json'
abstract property chain_type: ChainType

Return the chain type this handler belongs to.

Returns:

The ChainType enum value

abstract handle(request: Any, **kwargs) → Any[source]

Handle a request.

Parameters:
  • request – The request to handle

  • **kwargs – Additional keyword arguments

Returns:

The result of handling the request

abstract property handler_type: str

Return the handler type identifier.

This identifies the specific implementation (e.g., “json”, “md5”, “quicksort”).

Returns:

A string identifying the handler type

class pyiv.ChainType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Category key for a family of chain-of-responsibility handlers.

Why this exists: SerDe, network clients, and other pluggable handlers share the same registration/injection machinery but must not collide. ChainType namespaces those registries (encoding vs network, …).

Example

>>> from pyiv.chain import ChainType
>>> ChainType.ENCODING.value
'encoding'
>>> ChainType.NETWORK_CLIENT.value
'network_client'
ENCODING = 'encoding'
HASHING = 'hashing'
NETWORK_CLIENT = 'network_client'
SORTING = 'sorting'
class pyiv.Clock[source]

Bases: ABC

Abstract clock interface for time operations.

Why this exists: Call sites that use time.time()/sleep couple production to wall clock; inject Clock and bind SyntheticClock in tests.

This abstract class provides methods for time-related operations, allowing implementations to be swapped for testing or different time sources. Use RealClock for production code and SyntheticClock for testing.

Example

>>> from pyiv.clock import SyntheticClock
>>> clock = SyntheticClock(100.0)
>>> clock.time()
100.0
>>> clock.sleep(1.5)
>>> clock.time()
101.5
abstract monotonic() → float[source]

Get monotonic time (not affected by system clock adjustments).

Returns:

Monotonic time as float

abstract sleep(seconds: float) → None[source]

Sleep for specified duration.

Parameters:

seconds – Duration to sleep

abstract start_timer(interval: float, callback: Callable[[], None], repeat: bool = False) → Timer[source]

Start a timer that calls callback after interval.

Parameters:
  • interval – Time interval in seconds

  • callback – Function to call

  • repeat – If True, timer repeats

Returns:

Timer object

abstract thread_sleep(seconds: float) → None[source]

Sleep in current thread.

Parameters:

seconds – Duration to sleep

abstract time() → float[source]

Get current time as seconds since epoch.

Returns:

Current time as float

class pyiv.Command(args: Namespace, injector: Any | None = None)[source]

Bases: ABC

Base interface for discoverable CLI commands.

Use this when building hierarchical CLIs (command / subcommand trees) that should be discoverable via reflection and optionally receive a pyiv injector. Prefer CLICommand for one-shot tools and ServiceCommand for long-running processes.

Example

>>> import argparse
>>> from pyiv.command import Command
>>> class HelloCommand(Command):
...     @classmethod
...     def get_name(cls) -> str:
...         return "hello"
...     def execute(self) -> int:
...         return 0
>>> HelloCommand.get_name()
'hello'
>>> HelloCommand(argparse.Namespace()).execute()
0
__init__(args: Namespace, injector: Any | None = None)[source]

Initialize the command with parsed arguments.

Parameters:
  • args – Parsed command-line arguments

  • injector – Optional dependency injector for DI support

classmethod add_args(parser: ArgumentParser) → None[source]

Add arguments to the command’s argument parser.

Parameters:
  • cls – The class (implicit in classmethod)

  • parser – Argument parser to add arguments to

cleanup() → None[source]

Cleanup after command execution (optional lifecycle hook).

Override this method to: - Close database connections - Stop background tasks - Clean up any resources - Perform graceful shutdown

Called after execute() completes, even if an exception occurs.

abstract execute() → int[source]

Execute the command.

For long-running services, this typically calls: 1. init() - Initialize resources 2. run() - Main service loop 3. cleanup() - Clean up resources

For one-shot commands, implement the logic directly here.

Returns:

Exit code (0 for success, non-zero for error)

classmethod get_aliases() → List[str][source]

Get command aliases.

Parameters:

cls – The class (implicit in classmethod)

Returns:

List of alternative names for this command

classmethod get_description() → str[source]

Get the command description.

Parameters:

cls – The class (implicit in classmethod)

Returns:

Command description for help text

abstract classmethod get_name() → str[source]

Get the command name.

Parameters:

cls – The class (implicit in classmethod)

Returns:

Command name (e.g., “switchboard”, “start”, “create”)

classmethod get_subcommands() → List[Type[Command]][source]

Get subcommands of this command.

Parameters:

cls – The class (implicit in classmethod)

Returns:

List of command classes that are subcommands of this command

init() → None[source]

Initialize the command (optional lifecycle hook).

Override this method to: - Load configuration - Initialize resources (database connections, clients, etc.) - Set up signal handlers if needed - Perform any one-time setup

This is called automatically by execute() before run().

run() → None[source]

Run the command (optional lifecycle hook for long-running services).

Override this method for long-running services that need a main loop. For one-shot commands, override execute() directly instead.

This method should: - Start the main service loop - Block until service should stop - Handle the primary service functionality

This method should check self._shutdown_event periodically and exit gracefully when shutdown is requested.

setup_signal_handlers() → None[source]

Setup signal handlers for graceful shutdown.

Call this in init() for long-running services that need graceful shutdown.

class pyiv.CommandRunner(config: Any | None = None)[source]

Bases: object

Discovers and runs Command subclasses via reflection or import.

Use this as the entrypoint for a CLI package: discover commands under a package path, build an argparse tree, and execute the selected command with an optional pyiv injector from config.

Why this exists: Discover and dispatch hierarchical CLI commands with a shared lifecycle.

Example

>>> import argparse
>>> from pyiv.command import CLICommand, CommandRunner
>>> class GreetCommand(CLICommand):
...     @classmethod
...     def get_name(cls) -> str:
...         return "greet"
...     @classmethod
...     def get_description(cls) -> str:
...         return "Say hello"
...     def run(self) -> None:
...         self._exit_code = 0
>>> runner = CommandRunner()
>>> parser = runner.create_parser(
...     prog="demo", commands={"greet": GreetCommand}
... )
>>> args = parser.parse_args(["greet"])
>>> GreetCommand(args).execute()
0
__init__(config: Any | None = None)[source]

Initialize the command runner.

Parameters:

config – Optional pyiv Config instance for command discovery

create_parser(prog: str | None = None, description: str | None = None, commands: Dict[str, Type[Command]] | None = None) → ArgumentParser[source]

Create argument parser with discovered commands.

Parameters:
  • prog – Program name

  • description – Program description

  • commands – Optional pre-discovered commands dict

Returns:

Configured argument parser

discover_commands(package_path: str, pattern: str | None = None, recursive: bool = True) → Dict[str, Type[Command]][source]

Discover commands in a package using reflection.

Parameters:
  • package_path – Python package path (e.g., “agenticness.commands”)

  • pattern – Optional name pattern for filtering (e.g. *Command)

  • recursive – Whether to scan submodules recursively

Returns:

Dictionary mapping command names to command classes

run(package_path: str, prog: str | None = None, description: str | None = None, args: List[str] | None = None) → int[source]

Discover commands and run the appropriate one.

Parameters:
  • package_path – Python package path to discover commands in

  • prog – Program name

  • description – Program description

  • args – Optional command-line arguments (defaults to sys.argv[1:])

Returns:

Exit code from command execution

class pyiv.Config[source]

Bases: object

Module of bindings that describe how to build an object graph.

Why this exists: Without a registration surface, every call site must know concrete classes. Subclass Config, override configure(), and register interfaces → implementations (or use get_binder()).

Example

>>> from pyiv import Config, get_injector
>>> class Database:
...     pass
>>> class PostgreSQL(Database):
...     pass
>>> class MyConfig(Config):
...     def configure(self):
...         self.register(Database, PostgreSQL)
>>> isinstance(get_injector(MyConfig).inject(Database), PostgreSQL)
True
__init__()[source]

Initialize the configuration.

configure()[source]

Override this method to register dependencies.

Example

def configure(self):

self.register(AbstractClass, ConcreteClass) self.register_instance(Logger, my_logger_instance)

expose(type_or_key: Type | Key[Any]) → None[source]

Mark a binding as exposed from a private module.

On a regular Config this is a no-op store; PrivateConfig uses the set when installed into a parent.

get_binder() → Binder[source]

Get a binder for fluent configuration.

Returns:

A binder instance

get_chain_handler_instance(chain_type: ChainType, name: str) → ChainHandler | None[source]

Get a pre-registered chain handler instance.

Parameters:
  • chain_type – The chain type

  • name – The handler instance name

Returns:

The chain handler instance, or None if not found

get_chain_handler_registration(chain_type: ChainType, handler_type: str) → Type[ChainHandler] | None[source]

Get the registered chain handler class for a handler type.

Parameters:
  • chain_type – The chain type

  • handler_type – The handler type identifier

Returns:

The registered chain handler class, or None if not found

get_chain_handler_registration_by_name(chain_type: ChainType, name: str) → Tuple[Type[ChainHandler], str] | None[source]

Get the registered chain handler class and handler type for a name.

Parameters:
  • chain_type – The chain type

  • name – The handler instance name

Returns:

A tuple of (handler class, handler_type), or None if not found

get_chain_handler_singleton_type(chain_type: ChainType, name: str) → SingletonType[source]

Get the singleton type for a chain handler registration.

Parameters:
  • chain_type – The chain type

  • name – The handler instance name or handler type

Returns:

The singleton type, or SingletonType.NONE if not registered or not a singleton

get_exposed() → Set[Type | Key[Any]][source]

Return types/keys marked for exposure from a private module.

get_instance(abstract: Type) → Any | None[source]

Get a registered singleton instance for an abstract type.

Parameters:

abstract – The abstract class or interface

Returns:

The registered instance or None if not found

get_key_binding(key: Key[Any]) → Tuple[Type, Provider[Any] | None, Scope | None] | None[source]

Get a qualified binding for a key.

Parameters:

key – The qualified key

Returns:

Tuple of (type, provider, scope) or None if not found

get_map_multibinding(value_type: Type[T]) → Tuple[Dict[Any, Type], Dict[Any, Any]] | None[source]

Get map multibinding data for a value type.

Returns:

Tuple of (key->implementation, key->instance), or None if empty

get_multibinding(interface: Type[T]) → Tuple[Set[Type], List[Type], Set[Any], List[Any]] | None[source]

Get multibinding implementations for an interface.

Parameters:

interface – The interface type

Returns:

Tuple of (set_impls, list_impls, set_instances, list_instances) or None

get_provider(abstract: Type) → Provider[Any] | None[source]

Get the provider for a registered type.

Parameters:

abstract – The abstract class or interface

Returns:

The provider, or None if not registered or no provider

get_registration(abstract: Type) → Type | Callable | None[source]

Get the registered concrete implementation for an abstract type.

Parameters:

abstract – The abstract class or interface

Returns:

The registered concrete class, callable, or None if not found

get_scope(abstract: Type) → Scope | None[source]

Get the scope for a registered type.

Parameters:

abstract – The abstract class or interface

Returns:

The scope, or None if not registered or no scope

get_singleton_type(abstract: Type) → SingletonType[source]

Get the singleton type for a registered abstract type.

Parameters:

abstract – The abstract class or interface

Returns:

The singleton type, or SingletonType.NONE if not registered or not a singleton

has_chain_handler_registration(chain_type: ChainType, handler_type: str) → bool[source]

Check if a chain handler is registered for a handler type.

Parameters:
  • chain_type – The chain type

  • handler_type – The handler type identifier

Returns:

True if registered, False otherwise

has_chain_handler_registration_by_name(chain_type: ChainType, name: str) → bool[source]

Check if a chain handler is registered by name.

Parameters:
  • chain_type – The chain type

  • name – The handler instance name

Returns:

True if registered, False otherwise

has_registration(abstract: Type) → bool[source]

Check if a registration exists for an abstract type.

Parameters:

abstract – The abstract class or interface

Returns:

True if registered, False otherwise

install(other: Type[Config] | Config) → None[source]

Install another config module into this one.

Regular configs are merged immediately (last wins for the same type or key). PrivateConfig instances are queued and wired when the injector is created so exposed bindings can delegate into a child environment.

map_multibinder(value_type: Type[T]) → MapMultibinder[Any, T][source]

Create a map multibinder for keyed implementations of value_type.

merge_from(other: Config, *, replace: bool = True) → None[source]

Merge bindings from other into this config.

Parameters:
  • other – Source config

  • replace – If True, other overwrites existing keys; if False, existing bindings are kept.

multibinder(interface: Type[T], as_set: bool = True) → Multibinder[T][source]

Create a multibinder for multiple implementations.

Parameters:
  • interface – The interface type

  • as_set – If True, creates SetMultibinder, else ListMultibinder

Returns:

A multibinder instance

register(abstract: Type, concrete: Type | Callable, *, singleton: bool = False, singleton_type: SingletonType = SingletonType.NONE, scope: Scope | None = None, provider: Provider[Any] | None = None)[source]

Register a concrete implementation for an abstract type.

Parameters:
  • abstract – The abstract class or interface to register

  • concrete – The concrete class, instance, or factory function

  • singleton – If True, uses SINGLETON type (deprecated, use singleton_type or scope instead)

  • singleton_type – Type of singleton behavior (NONE, SINGLETON, or GLOBAL_SINGLETON)

  • scope – Scope for lifecycle management (takes precedence over singleton_type)

  • provider – Provider to use for instance creation (takes precedence over concrete)

Raises:
  • TypeError – If abstract is not a type

  • ValueError – If conflicting parameters are specified

register_chain_handler(chain_type: ChainType, handler_type: str, handler_class: Type[ChainHandler], *, singleton_type: SingletonType = SingletonType.SINGLETON)[source]

Register a chain handler implementation for a handler type.

This registers a default implementation for the handler type. When injecting by handler type (without a specific name), this implementation will be used.

Parameters:
  • chain_type – The chain type (e.g., ChainType.ENCODING, ChainType.HASHING)

  • handler_type – The handler type identifier (e.g., “json”, “md5”, “quicksort”)

  • handler_class – The chain handler implementation class

  • singleton_type – Type of singleton behavior (default: SINGLETON)

Raises:
  • TypeError – If handler_class is not a subclass of ChainHandler

  • ValueError – If handler_type is empty

register_chain_handler_by_name(chain_type: ChainType, name: str, handler_class: Type[ChainHandler], handler_type: str, *, singleton_type: SingletonType = SingletonType.SINGLETON)[source]

Register a named chain handler implementation.

This allows multiple implementations of the same handler type with different behaviors (e.g., “json-input”, “json-output”, “md5-fast”, “md5-secure”).

Parameters:
  • chain_type – The chain type (e.g., ChainType.ENCODING, ChainType.HASHING)

  • name – Unique name for this handler instance

  • handler_class – The chain handler implementation class

  • handler_type – The handler type identifier (e.g., “json”, “md5”)

  • singleton_type – Type of singleton behavior (default: SINGLETON)

Raises:
  • TypeError – If handler_class is not a subclass of ChainHandler

  • ValueError – If name or handler_type is empty

register_chain_handler_instance(chain_type: ChainType, name: str, instance: ChainHandler)[source]

Register a pre-created chain handler instance.

Parameters:
  • chain_type – The chain type

  • name – Unique name for this handler instance

  • instance – The pre-created chain handler instance

Raises:
register_instance(abstract: Type, instance: Any)[source]

Register a concrete instance for an abstract type.

Parameters:
  • abstract – The abstract class or interface

  • instance – The concrete instance to register

register_key(key: Key[Any], implementation: Type | Provider[Any], *, scope: Scope | None = None) → None[source]

Register a qualified binding using a Key.

Parameters:
  • key – The qualified key

  • implementation – The implementation class or provider

  • scope – Optional scope for lifecycle management

register_map_multibinding(value_type: Type[T], key: Any, implementation: Type[T]) → None[source]

Register a keyed implementation in a map multibinding.

register_map_multibinding_instance(value_type: Type[T], key: Any, instance: T) → None[source]

Register a keyed instance in a map multibinding.

register_multibinding(interface: Type[T], implementation: Type[T], *, as_set: bool = True) → None[source]

Register an implementation in a multibinding.

Parameters:
  • interface – The interface type

  • implementation – The implementation class

  • as_set – If True, adds to set, else to list

register_multibinding_instance(interface: Type[T], instance: T, *, as_set: bool = True) → None[source]

Register a pre-created instance in a multibinding.

Parameters:
  • interface – The interface type

  • instance – The instance to add

  • as_set – If True, adds to set, else to list

register_provider(abstract: Type, provider: Provider[Any]) → None[source]

Register a provider for a type.

Parameters:
  • abstract – The abstract class or interface

  • provider – The provider to use for instance creation

require_explicit_bindings() → None[source]

Disable just-in-time construction of unregistered concrete types.

requires_explicit_bindings() → bool[source]

Return whether explicit bindings are required.

singleton_bindings() → List[Type][source]

Return types that should be eagerly created in Stage.PRODUCTION.

class pyiv.Console(*args, **kwargs)[source]

Bases: Protocol

Protocol for console output implementations.

Why this exists: print() to sys.stdout is hard to assert; inject Console and capture with MemoryConsole in tests.

Console provides a file-like interface for output, allowing print() statements to use dependency injection. This enables testing by capturing output without actually printing to stdout.

Console implementations should be file-like objects that support the standard file interface (write, flush, etc.) used by print().

Console also includes terminal functionality (TTY operations) for advanced features like spinners, animations, password prompts, and progress bars. Implementations that don’t support terminal features should provide no-op implementations that return safe defaults.

Example

>>> from pyiv.console import MemoryConsole
>>>
>>> console = MemoryConsole()
>>> print("Hello, World!", file=console)
>>> console.flush()
>>> "Hello, World!" in console.getvalue()
True
__init__(*args, **kwargs)
bold(enabled: bool = True) → None[source]

Enable or disable bold text.

clear() → None[source]

Clear the entire screen.

clear_line() → None[source]

Clear the current line.

clear_state() → None[source]

Clear all state (for test setup).

Resets terminal state including screen buffer, cursor position, colors, styles, and event history.

clear_to_end_of_line() → None[source]

Clear from cursor to end of line.

flush() → None[source]

Flush any buffered output.

Ensures all written data is actually output.

get_color() → Tuple[int | None, int | None][source]

Get current color state (for state inspection).

Returns:

Tuple of (fg, bg)

get_cursor() → Tuple[int, int][source]

Get current cursor position (for state inspection).

Returns:

Tuple of (x, y)

get_cursor_position() → Tuple[int, int] | None[source]

Get current cursor position.

Returns:

Tuple of (x, y) or None if not available

get_echo_enabled() → bool[source]

Get echo state (for state inspection).

Returns:

True if echo is enabled

get_events() → List[TerminalEvent][source]

Get event history (for state inspection).

Returns:

List of terminal events

get_raw_mode() → bool[source]

Get raw mode state (for state inspection).

Returns:

True if raw mode is enabled

get_screen() → List[List[str]][source]

Get screen buffer (for state inspection).

Returns:

2D list of screen contents

get_screen_char(x: int, y: int) → str[source]

Get character at position from screen buffer.

Parameters:
  • x – Column (0-based)

  • y – Row (0-based)

Returns:

Character at position

get_screen_line(y: int) → str[source]

Get specific line from screen buffer.

Parameters:

y – Line number (0-based)

Returns:

Line contents as string

get_size() → Tuple[int, int][source]

Get terminal size.

Returns:

Tuple of (columns, rows). Defaults to (80, 24) if not available.

get_style() → Dict[str, bool][source]

Get text style state (for state inspection).

Returns:

Dictionary of style flags (bold, underline, etc.)

hide_cursor() → None[source]

Hide the cursor.

is_tty() → bool[source]

Check if output is a real terminal.

Returns:

True if output is connected to a TTY, False otherwise

move_cursor(x: int, y: int) → None[source]

Move cursor to position.

Parameters:
  • x – Column (0-based)

  • y – Row (0-based)

move_cursor_down(n: int = 1) → None[source]

Move cursor down n lines.

move_cursor_home() → None[source]

Move cursor to home position (0, 0).

move_cursor_left(n: int = 1) → None[source]

Move cursor left n columns.

move_cursor_right(n: int = 1) → None[source]

Move cursor right n columns.

move_cursor_up(n: int = 1) → None[source]

Move cursor up n lines.

read_char(timeout: float | None = None) → str | None[source]

Read a single character (requires raw mode).

Parameters:

timeout – Optional timeout in seconds

Returns:

Character string or None if no input available

read_line(prompt: str = '') → str[source]

Read a line of input.

Parameters:

prompt – Optional prompt string to display

Returns:

Line of text (without newline)

read_password(prompt: str = '') → str[source]

Read password with echo disabled.

Parameters:

prompt – Prompt string to display

Returns:

Password string

reset_color() → None[source]

Reset colors to default.

restore_cursor() → None[source]

Restore saved cursor position.

save_cursor() → None[source]

Save current cursor position.

set_color(fg: int | None = None, bg: int | None = None) → None[source]

Set foreground and/or background color.

Parameters:
  • fg – Foreground color code (30-37 for basic, 30-37,90-97 for bright)

  • bg – Background color code (40-47 for basic, 100-107 for bright)

set_echo(enabled: bool) → None[source]

Turn echo on/off for password prompts.

Parameters:

enabled – True to enable echo, False to disable

set_raw_mode(enabled: bool) → None[source]

Enable/disable raw mode for character-by-character input.

Parameters:

enabled – True for raw mode, False for cooked mode

show_cursor() → None[source]

Show the cursor.

underline(enabled: bool = True) → None[source]

Enable or disable underline.

writable() → bool[source]

Check if console is writable.

Returns:

True if console supports writing

write(s: str) → int[source]

Write string to console.

Parameters:

s – String to write

Returns:

Number of characters written

exception pyiv.CreationError(message: str, *, path: Sequence[str] | None = None, causes: Sequence[BaseException] | None = None)[source]

Bases: Exception

Raised when the injector cannot create or resolve a dependency.

Why this exists: Deep graphs fail far from the call site. This error carries the resolution path and optional nested causes so you can see App -> Service -> Database instead of a bare ValueError.

Parameters:
  • message – Human-readable failure reason

  • path – Resolution path from the root request to the failing type

  • causes – Nested failures (e.g. eager singleton warmup aggregation)

Example

>>> err = CreationError("No binding", path=["Service", "Database"])
>>> "Service -> Database" in str(err)
True
__init__(message: str, *, path: Sequence[str] | None = None, causes: Sequence[BaseException] | None = None)[source]
class pyiv.DateTimeService[source]

Bases: ABC

Abstract interface for UTC datetime operations.

Depend on this instead of calling datetime.now directly so production code can use PythonDateTimeService while tests inject MockDateTimeService with a fixed clock.

Example

>>> from datetime import datetime, timezone
>>> from pyiv.datetime_service import DateTimeService, MockDateTimeService
>>> def stamp(svc: DateTimeService) -> str:
...     return svc.now_utc_iso()
>>> fixed = datetime(2024, 1, 15, 10, 30, 0, tzinfo=timezone.utc)
>>> stamp(MockDateTimeService(fixed))
'2024-01-15T10:30:00+00:00'
abstract now_utc() → datetime[source]

Get current UTC datetime.

Returns:

Current datetime in UTC timezone

abstract now_utc_iso() → str[source]

Get current UTC datetime as ISO format string.

Returns:

30:00.123456+00:00”)

Return type:

Current datetime in UTC as ISO format string (e.g., “2024-01-15T10

class pyiv.Factory(*args, **kwargs)[source]

Bases: Protocol, Generic[T]

Callable creation API for objects that need runtime arguments.

Why this exists: A DI Provider builds one graph-managed instance. A Factory creates many instances with caller-supplied args (user id, connection string) while still allowing the factory itself to be injected.

Example

>>> from pyiv.factory import SimpleFactory
>>> class User:
...     def __init__(self, name: str):
...         self.name = name
>>> factory: Factory[User] = SimpleFactory(User)
>>> factory.create(name="Ada").name
'Ada'
__init__(*args, **kwargs)
create(*args: Any, **kwargs: Any) → T[source]

Create an instance of type T.

Parameters:
  • *args – Positional arguments for instance creation

  • **kwargs – Keyword arguments for instance creation

Returns:

An instance of type T

class pyiv.FactoryProvider(factory: Callable[[...], T])[source]

Bases: Generic[T]

Provider that wraps a factory function.

Why this exists: Adapt a zero-arg factory callable into the Provider protocol.

This provider calls a factory function each time get() is called. Useful for creating new instances on demand.

Example

>>> from pyiv.provider import FactoryProvider
>>>
>>> class User:
...     def __init__(self, name: str = "default"):
...         self.name = name
>>>
>>> def create_user() -> User:
...     return User(name="default")
>>>
>>> user_provider = FactoryProvider(create_user)
>>> user = user_provider.get()  # Calls create_user()
>>> user.name
'default'
>>> # Each call creates a new instance
>>> user2 = user_provider.get()
>>> user is not user2
True
__init__(factory: Callable[[...], T])[source]

Initialize provider with a factory function.

Parameters:

factory – A callable that creates instances of type T

get() → T[source]

Get an instance by calling the factory function.

Returns:

An instance created by the factory function

class pyiv.FileConsole(file: str | Path, mode: str = 'w', encoding: str = 'utf-8')[source]

Bases: BaseConsole

File-based console for testing.

Why this exists: When a test must verify output landed in a real file path.

This console writes output to a file, useful for testing scenarios where you want to verify output was written to a specific file.

Example

>>> import os
>>> import tempfile
>>> from pathlib import Path
>>> from pyiv.console import FileConsole
>>>
>>> fd, path = tempfile.mkstemp(suffix=".txt")
>>> os.close(fd)
>>> os.unlink(path)
>>> console = FileConsole(path)
>>> print("Test message", file=console)
>>> console.flush()
>>> "Test message" in Path(path).read_text()
True
>>> os.unlink(path)
__init__(file: str | Path, mode: str = 'w', encoding: str = 'utf-8')[source]

Initialize file-based console.

Parameters:
  • file – File path to write to

  • mode – File mode (default: “w” for write)

  • encoding – Text encoding (default: “utf-8”)

close() → None[source]

Close the file stream.

flush() → None[source]

Flush file buffer.

writable() → bool[source]

Check if file is writable.

Returns:

True if file mode supports writing

write(s: str) → int[source]

Write string to file.

Parameters:

s – String to write

Returns:

Number of characters written

class pyiv.Filesystem[source]

Bases: ABC

Abstract filesystem for injectable file I/O.

Why this exists: Production code that calls open() / pathlib directly is hard to unit-test without touching disk or mocking builtins. Depend on Filesystem and bind RealFilesystem in production and MemoryFilesystem in tests.

Example

>>> from pyiv.filesystem import MemoryFilesystem
>>> fs: Filesystem = MemoryFilesystem()
>>> fs.write_text("notes.txt", "hello")
>>> fs.read_text("notes.txt")
'hello'
abstract copy(src: str | Path, dst: str | Path) → None[source]

Copy a file.

Parameters:
  • src – Source path

  • dst – Destination path

abstract exists(path: str | Path) → bool[source]

Check if a path exists.

Parameters:

path – Path to check

Returns:

True if path exists, False otherwise

abstract get_size(path: str | Path) → int[source]

Get file size in bytes.

Parameters:

path – File path

Returns:

File size in bytes

abstract glob(pattern: str | Path) → Iterator[Path][source]

Glob pattern matching.

Parameters:

pattern – Glob pattern

Yields:

Matching paths

abstract is_dir(path: str | Path) → bool[source]

Check if path is a directory.

Parameters:

path – Path to check

Returns:

True if path is a directory, False otherwise

abstract is_file(path: str | Path) → bool[source]

Check if path is a file.

Parameters:

path – Path to check

Returns:

True if path is a file, False otherwise

abstract listdir(path: str | Path) → Iterator[str][source]

List directory contents.

Parameters:

path – Directory path

Yields:

Directory entry names

abstract mkdir(path: str | Path, parents: bool = False, exist_ok: bool = False) → None[source]

Create a directory.

Parameters:
  • path – Directory path

  • parents – Create parent directories if needed

  • exist_ok – Don’t raise error if directory exists

abstract move(src: str | Path, dst: str | Path) → None[source]

Move/rename a file or directory.

Parameters:
  • src – Source path

  • dst – Destination path

abstract open(file: str | Path, mode: str = 'r', encoding: str | None = None) → TextIO | BinaryIO[source]

Open a file.

Parameters:
  • file – File path

  • mode – File mode (r, w, a, rb, wb, etc.)

  • encoding – Text encoding (for text modes)

Returns:

File handle

abstract read_bytes(path: str | Path) → bytes[source]

Read bytes from a file.

Parameters:

path – File path

Returns:

File contents as bytes

abstract read_text(path: str | Path, encoding: str = 'utf-8') → str[source]

Read text from a file.

Parameters:
  • path – File path

  • encoding – Text encoding

Returns:

File contents as string

abstract rmdir(path: str | Path) → None[source]

Remove a directory.

Parameters:

path – Directory path

Remove a file.

Parameters:
  • path – File path

  • missing_ok – Don’t raise error if file doesn’t exist

abstract write_bytes(path: str | Path, content: bytes) → None[source]

Write bytes to a file.

Parameters:
  • path – File path

  • content – Bytes content to write

abstract write_text(path: str | Path, content: str, encoding: str = 'utf-8') → None[source]

Write text to a file.

Parameters:
  • path – File path

  • content – Text content to write

  • encoding – Text encoding

class pyiv.GlobalSingletonRegistry[source]

Bases: object

Process-wide store for GLOBAL_SINGLETON instances.

Why this exists: Some resources must be unique across injectors (shared caches). Tests call clear() between cases so state does not leak.

Example

>>> class Cache:
...     pass
>>> GlobalSingletonRegistry.clear()
>>> GlobalSingletonRegistry.set(Cache, Cache())
>>> GlobalSingletonRegistry.has(Cache)
True
>>> GlobalSingletonRegistry.clear()
classmethod clear() → None[source]

Clear all global singletons (useful for testing).

Parameters:

cls – The class (implicit in classmethod)

classmethod get(key: Type | str) → Any[source]

Get a global singleton instance.

Parameters:
  • cls – The class (implicit in classmethod)

  • key – The abstract type or string key to retrieve

Returns:

The singleton instance or None if not registered

classmethod has(key: Type | str) → bool[source]

Check if a global singleton exists.

Parameters:
  • cls – The class (implicit in classmethod)

  • key – The abstract type or string key to check

Returns:

True if a singleton exists, False otherwise

classmethod set(key: Type | str, instance: Any) → None[source]

Set a global singleton instance.

Parameters:
  • cls – The class (implicit in classmethod)

  • key – The abstract type or string key

  • instance – The instance to store

class pyiv.GlobalSingletonScope[source]

Bases: object

Global singleton scope (thread-safe).

Why this exists: One instance shared by all injectors in the process (thread-safe).

This scope caches instances globally across all injectors. The same instance is shared by all injectors and all threads. Access is thread-safe.

This is useful for truly global singletons like configuration, caches, or shared resources.

Example

>>> from pyiv.scope import GlobalSingletonScope
>>> from pyiv import Config
>>>
>>> class MyConfig(Config):
...     def configure(self):
...         # Global singleton (shared across all injectors)
...         self.register(Cache, RedisCache, scope=GlobalSingletonScope())
classmethod clear() → None[source]

Clear all global singleton instances (useful for testing).

Parameters:

cls – The class (implicit in classmethod)

classmethod has(key: Type | str | tuple) → bool[source]

Check if a global singleton exists.

Parameters:
  • cls – The class (implicit in classmethod)

  • key – The key to check

Returns:

True if a singleton exists, False otherwise

scope(key: Type | str | tuple, provider: Provider[Any]) → Provider[Any][source]

Scope provider to global singleton.

Parameters:
  • key – The key identifying the dependency

  • provider – The provider to scope

Returns:

A provider that returns the same global instance (thread-safe)

class pyiv.HTTPClient[source]

Bases: NetworkClient

HTTP(S) client using stdlib urllib.

Use this for plain HTTP or HTTPS URLs when you want a zero-dependency client bound into the NETWORK_CLIENT chain. Prefer invalid-URL Traceback examples in docs/tests — do not hit the network from doctests.

Example

>>> client = HTTPClient()
>>> client.handler_type
'http'
>>> client.request("GET", "ftp://example.com")
Traceback (most recent call last):
    ...
ValueError: HTTPClient only supports http:// and https:// URLs, got: ftp://example.com
property handler_type: str

Return the handler type identifier.

Returns:

The handler type identifier (“http”)

request(method: str, url: str, headers: Dict[str, str] | None = None, data: str | bytes | None = None, timeout: float | None = None) → Dict[str, Any][source]

Make an HTTP request.

Parameters:
  • method – HTTP method (GET, POST, PUT, DELETE, etc.)

  • url – The URL to request (must start with http://)

  • headers – Optional dictionary of HTTP headers

  • data – Optional request body (string or bytes)

  • timeout – Optional timeout in seconds

Returns:

  • status: HTTP status code

  • headers: Response headers dictionary

  • body: Response body (bytes)

  • url: Final URL after redirects

Return type:

Dictionary containing

Raises:
  • ValueError – If URL does not start with http://

  • URLError – If the request fails

class pyiv.HTTPSClient[source]

Bases: NetworkClient

HTTPS-only client using stdlib urllib.

Use this when you want to reject non-HTTPS URLs at the client boundary (scheme must be https://). Like HTTPClient, doctests should use Traceback examples for invalid schemes rather than live network calls.

Example

>>> client = HTTPSClient()
>>> client.handler_type
'https'
>>> client.request("GET", "http://example.com")
Traceback (most recent call last):
    ...
ValueError: HTTPSClient only supports https:// URLs, got: http://example.com
property handler_type: str

Return the handler type identifier.

Returns:

The handler type identifier (“https”)

request(method: str, url: str, headers: Dict[str, str] | None = None, data: str | bytes | None = None, timeout: float | None = None) → Dict[str, Any][source]

Make an HTTPS request.

Parameters:
  • method – HTTP method (GET, POST, PUT, DELETE, etc.)

  • url – The URL to request (must start with https://)

  • headers – Optional dictionary of HTTP headers

  • data – Optional request body (string or bytes)

  • timeout – Optional timeout in seconds

Returns:

  • status: HTTP status code

  • headers: Response headers dictionary

  • body: Response body (bytes)

  • url: Final URL after redirects

Return type:

Dictionary containing

Raises:
class pyiv.Injector(config: Config, *, stage: Stage = Stage.DEVELOPMENT, parent: Injector | None = None, wire_private: bool = True)[source]

Bases: object

Resolves types and keys from a Config graph.

Why this exists: Manual wiring (new / factories everywhere) couples construction to call sites. The injector builds objects from bindings and constructor annotations so you register once and request by type.

Example

>>> from pyiv import Config, get_injector
>>> class Database:
...     pass
>>> class PostgreSQL(Database):
...     pass
>>> class MyConfig(Config):
...     def configure(self):
...         self.register(Database, PostgreSQL)
>>> injector = get_injector(MyConfig)
>>> isinstance(injector.inject(Database), PostgreSQL)
True
__init__(config: Config, *, stage: Stage = Stage.DEVELOPMENT, parent: Injector | None = None, wire_private: bool = True)[source]

Initialize the injector with a configuration.

Parameters:
  • config – The configuration object that defines dependencies

  • stage – DEVELOPMENT (lazy) or PRODUCTION (eager singletons)

  • parent – Optional parent injector for hierarchical lookup

  • wire_private – If True, install pending PrivateConfig modules

create_child(config: Type[Config] | Config) → Injector[source]

Create a child injector that inherits parent bindings.

The child sees parent bindings on miss; the parent cannot see child bindings. Singletons created in the child are local to the child.

Example

>>> from pyiv import Config, get_injector
>>> class Database:
...     pass
>>> class ProdConfig(Config):
...     def configure(self):
...         self.register(Database, Database)
>>> class RequestId:
...     def __init__(self, value: str = "r1"):
...         self.value = value
>>> class RequestConfig(Config):
...     def configure(self):
...         self.register(RequestId, RequestId)
>>> root = get_injector(ProdConfig)
>>> child = root.create_child(RequestConfig)
>>> isinstance(child.inject(Database), Database)
True
>>> child.inject(RequestId).value
'r1'
eager_singletons() → None[source]

Create all singleton-scoped bindings now (used by Stage.PRODUCTION).

inject(cls_or_key: Type | Key[Any], **kwargs) → Any[source]

Inject and create an instance of the given class or key.

Parameters:
  • cls_or_key – The class to instantiate (can be abstract or concrete) or a Key

  • **kwargs – Additional keyword arguments to pass to the constructor

Returns:

An instance of the class (or registered concrete implementation)

Raises:

CreationError – If the binding cannot be resolved (with path context)

inject_by_name(interface: Type, name: str) → Type[source]

Inject a specific implementation by name (ReflectionConfig).

inject_chain_handler(chain_type: ChainType, handler_type: str) → ChainHandler[source]

Inject a chain handler instance by handler type.

inject_chain_handler_by_name(chain_type: ChainType, name: str) → ChainHandler[source]

Inject a chain handler instance by name.

inject_members(instance: Any) → None[source]

Inject dependencies into an existing instance.

property parent: Injector | None

Return the parent injector, if any.

property stage: Stage

Return the stage this injector was created with.

class pyiv.InjectorMembersInjector(cls: Type[T], injector: Any)[source]

Bases: Generic[T]

MembersInjector that fills fields on instances you already constructed.

Why this exists: Frameworks and legacy code often create objects outside the injector. Use this (or injector.inject_members) when constructor injection is impractical—dataclasses, third-party types, or migration.

Supports field injection (dataclasses, attrs, regular classes) and method injection for annotated methods.

Example

>>> from pyiv.members import InjectorMembersInjector
>>> from dataclasses import dataclass, field
>>> from pyiv import Config, get_injector
>>>
>>> class Database:
...     pass
>>>
>>> @dataclass
... class Service:
...     db: Database = field(default=None)
>>>
>>> class MyConfig(Config):
...     def configure(self):
...         self.register(Database, Database)
>>>
>>> injector = get_injector(MyConfig)
>>> members_injector = InjectorMembersInjector(Service, injector)
>>> service = Service()
>>> members_injector.inject_members(service)
>>> isinstance(service.db, Database)
True
__init__(cls: Type[T], injector: Any)[source]

Initialize members injector.

Parameters:
  • cls – The class type to inject into

  • injector – The injector to use for dependency resolution

inject_members(instance: T) → None[source]

Inject dependencies into an existing instance.

This method: 1. Inspects the class for fields with type annotations 2. Resolves dependencies using the injector 3. Sets field values on the instance 4. Optionally calls methods with injected dependencies

Parameters:

instance – The instance to inject dependencies into

Raises:

TypeError – If instance is not of the expected type

class pyiv.InjectorProvider(cls: Type[T], injector: Any)[source]

Bases: Generic[T]

Provider that uses an injector to create instances.

This provider wraps an injector and a type, delegating instance creation to the injector. This is useful when you need a Provider interface but want to use the injector’s full dependency resolution.

Why this exists: Provider that asks an Injector for T on each get() — lazy graph lookup.

Example

>>> from pyiv import Config, get_injector
>>> from pyiv.provider import InjectorProvider
>>>
>>> class Database:
...     pass
>>>
>>> class MyConfig(Config):
...     def configure(self):
...         self.register(Database, Database)
>>>
>>> injector = get_injector(MyConfig)
>>> db_provider = InjectorProvider(Database, injector)
>>> db = db_provider.get()  # Uses injector.inject(Database)
>>> isinstance(db, Database)
True
__init__(cls: Type[T], injector: Any)[source]

Initialize provider with a type and injector.

Parameters:
  • cls – The type to provide instances of

  • injector – The injector to use for instance creation

get() → T[source]

Get an instance using the injector.

Returns:

An instance of type T created by the injector

class pyiv.InstanceProvider(instance: T)[source]

Bases: Generic[T]

Provider that returns a pre-created instance.

Why this exists: Wrap an already-built object as a Provider for binder.to_provider / register_provider.

This provider simply returns the same instance every time get() is called. Useful for wrapping pre-created singletons or instances.

Example

>>> from pyiv.provider import InstanceProvider
>>>
>>> class Logger:
...     def __init__(self, name: str):
...         self.name = name
>>>
>>> my_logger = Logger("app")
>>> logger_provider = InstanceProvider(my_logger)
>>> logger = logger_provider.get()  # Returns my_logger
>>> logger is my_logger
True
>>> logger.name
'app'
__init__(instance: T)[source]

Initialize provider with a pre-created instance.

Parameters:

instance – The instance to return

get() → T[source]

Get the pre-created instance.

Returns:

The instance that was provided during initialization

class pyiv.JSONSerDe[source]

Bases: SerDe

Standard JSON SerDe using Python’s json module.

Use this for APIs, configs, and payloads that exchange JSON. Datetimes serialize to ISO strings; objects with __dict__ become dicts. Prefer this over pickle whenever the data is text-safe and interoperable.

Example

>>> from pyiv.serde.encodings import JSONSerDe
>>> serde = JSONSerDe()
>>> serde.handler_type
'json'
>>> serde.serialize({"a": 1})
'{"a": 1}'
>>> serde.deserialize('{"a": 1}')
{'a': 1}
deserialize(data: str | bytes, target_type: Type[T] | None = None) → T[source]

Deserialize JSON string/bytes back to a Python object.

Parameters:
  • data – The JSON string or bytes

  • target_type – Optional type hint for the expected result type

Returns:

Deserialized Python object

property handler_type: str

Return the handler type identifier.

Returns:

The handler type identifier (“json”)

serialize(obj: Any) → str[source]

Serialize using standard JSON encoding.

Parameters:

obj – The Python object to serialize

Returns:

JSON string representation

class pyiv.Key(binding_type: Type[T], qualifier: Qualifier | None = None)[source]

Bases: Generic[T]

Type-safe key for qualified bindings.

Why this exists: Pair a type with an optional qualifier so multiple implementations of one type can coexist.

A Key combines a type with an optional qualifier to create a unique binding key. This allows multiple implementations of the same type to be registered and injected.

Example

>>> from pyiv.key import Key, Named
>>>
>>> # Key without qualifier (default binding)
>>> default_key = Key(int)
>>>
>>> # Key with qualifier
>>> primary_key = Key(int, Named("primary"))
>>> replica_key = Key(int, Named("replica"))
>>> primary_key != replica_key
True
__init__(binding_type: Type[T], qualifier: Qualifier | None = None)[source]

Initialize a key.

Parameters:
  • binding_type – The type to bind

  • qualifier – Optional qualifier to distinguish this binding

Raises:

TypeError – If binding_type is not a type

class pyiv.ListMultibinder(interface: Type[T], config: Any)[source]

Bases: Generic[T]

Multibinder that binds to a List[T].

Why this exists: Collect implementations as List[T] preserving add order.

This multibinder collects implementations into a list, preserving order. Duplicates are allowed.

Example

>>> from pyiv import Config
>>>
>>> class Validator:
...     pass
>>> class EmailValidator(Validator):
...     pass
>>> class MyConfig(Config):
...     def configure(self):
...         pass
>>> multibinder = ListMultibinder(Validator, MyConfig())
>>> multibinder.add(EmailValidator)
>>> EmailValidator in multibinder.get_implementations()
True
__init__(interface: Type[T], config: Any)[source]

Initialize list multibinder.

Parameters:
  • interface – The interface type

  • config – The config to register bindings with

add(implementation: Type[T]) → None[source]

Add an implementation class.

Parameters:

implementation – The implementation class to add

add_instance(instance: T) → None[source]

Add a pre-created instance.

Parameters:

instance – The instance to add

get_implementations() → List[Type[T]][source]

Get all registered implementation classes.

Returns:

List of implementation classes (order preserved)

get_instances() → List[T][source]

Get all registered instances.

Returns:

List of instances (order preserved)

class pyiv.MapMultibinder(value_type: Type[V], config: Any)[source]

Bases: Generic[K, V]

Multibinder that binds to a Dict[K, V].

Why this exists: Collect keyed implementations for Dict[K, V] injection.

Collects keyed implementations for injection as a mapping. Inject the dict through a host class constructor, not injector.inject(Dict[...]).

Example

>>> from typing import Dict
>>> from pyiv import Config, get_injector
>>> class Plugin:
...     pass
>>> class AuthPlugin(Plugin):
...     pass
>>> class MyConfig(Config):
...     def configure(self):
...         self.map_multibinder(Plugin).add("auth", AuthPlugin)
>>> class Host:
...     def __init__(self, plugins: Dict[str, Plugin]):
...         self.plugins = plugins
>>> "auth" in get_injector(MyConfig).inject(Host).plugins
True
__init__(value_type: Type[V], config: Any)[source]
add(key: K, implementation: Type[V]) → None[source]

Add a keyed implementation class.

add_instance(key: K, instance: V) → None[source]

Add a keyed pre-created instance.

get_implementations() → Dict[K, Type[V]][source]

Get all registered keyed implementation classes.

get_instances() → Dict[K, V][source]

Get all registered keyed instances.

class pyiv.MembersInjector(*args, **kwargs)[source]

Bases: Protocol, Generic[T]

Protocol for injecting dependencies into existing instances.

Why this exists: Inject into objects you did not construct (frameworks, dataclasses, legacy).

MembersInjectors inject dependencies into fields and methods of existing instances. This is useful for: - Framework integration (Django, Flask, etc.) - Legacy code migration - Third-party object injection - Field injection for data classes

Example

class MyMembersInjector(MembersInjector[Service]):
def inject_members(self, instance: Service) -> None:

instance.db = self._injector.inject(Database)

__init__(*args, **kwargs)
inject_members(instance: T) → None[source]

Inject dependencies into an existing instance.

Parameters:

instance – The instance to inject dependencies into

class pyiv.MemoryConsole[source]

Bases: BaseConsole

In-memory console for testing.

Why this exists: Capture print(…, file=console) without touching stdout.

This console stores output in memory, allowing tests to capture and verify console output without actually printing to stdout.

Example

>>> from pyiv.console import MemoryConsole
>>>
>>> console = MemoryConsole()
>>> print("Test message", file=console)
>>> print("Another line", file=console)
>>>
>>> # Get all captured output
>>> output = console.getvalue()
>>> assert "Test message" in output
>>> assert "Another line" in output
>>>
>>> # Clear and reuse
>>> console.seek(0)
0
>>> console.truncate(0)
0
>>> print("New output", file=console)
>>> console.getvalue()
'New output\n'
__init__()[source]

Initialize in-memory console.

close() → None[source]

Close the buffer.

Note: After closing, getvalue() will raise ValueError. To get the value before closing, call getvalue() first.

flush() → None[source]

Flush buffer (no-op for memory console).

getvalue() → str[source]

Get all captured output.

Returns:

All output written to the console as a string

seek(pos: int) → int[source]

Seek to position in buffer.

Parameters:

pos – Position to seek to

Returns:

New position

truncate(size: int | None = None) → int[source]

Truncate buffer to size.

Parameters:

size – Size to truncate to (None = current position)

Returns:

New size

writable() → bool[source]

Check if console is writable.

Returns:

True (memory console is always writable)

write(s: str) → int[source]

Write string to memory buffer.

Parameters:

s – String to write

Returns:

Number of characters written

class pyiv.MemoryFilesystem[source]

Bases: Filesystem

In-memory filesystem for testing.

Use this whenever code under test needs file I/O but must not touch the real disk. Paths live in a dict-backed store, so tests stay fast and isolated.

Example

>>> from pyiv.filesystem import MemoryFilesystem
>>> fs = MemoryFilesystem()
>>> fs.write_text("/notes/a.txt", "alpha")
>>> fs.mkdir("/notes/sub", parents=True, exist_ok=True)
>>> fs.exists("/notes/a.txt")
True
>>> fs.read_text("/notes/a.txt")
'alpha'
__init__()[source]

Initialize in-memory filesystem.

copy(src: str | Path, dst: str | Path) → None[source]

Copy a file.

Parameters:
  • src – Source file path

  • dst – Destination file path

Raises:

FileNotFoundError – If source file doesn’t exist

exists(path: str | Path) → bool[source]

Check if path exists.

Parameters:

path – Path to check

Returns:

True if path exists (as file or directory), False otherwise

get_size(path: str | Path) → int[source]

Get file size in bytes.

Parameters:

path – File path

Returns:

File size in bytes

Raises:

FileNotFoundError – If file doesn’t exist

glob(pattern: str | Path) → Iterator[Path][source]

Glob pattern matching (simplified).

Parameters:

pattern – Glob pattern to match

Yields:

Matching paths

is_dir(path: str | Path) → bool[source]

Check if path is a directory.

Parameters:

path – Path to check

Returns:

True if path is a directory, False otherwise

is_file(path: str | Path) → bool[source]

Check if path is a file.

Parameters:

path – Path to check

Returns:

True if path is a file, False otherwise

listdir(path: str | Path) → Iterator[str][source]

List directory contents.

Parameters:

path – Directory path to list

Yields:

Directory entry names

Raises:

FileNotFoundError – If directory doesn’t exist

mkdir(path: str | Path, parents: bool = False, exist_ok: bool = False) → None[source]

Create a directory.

Parameters:
  • path – Directory path to create

  • parents – Create parent directories if needed

  • exist_ok – Don’t raise error if directory already exists

Raises:
move(src: str | Path, dst: str | Path) → None[source]

Move/rename a file or directory.

Parameters:
  • src – Source path

  • dst – Destination path

Raises:

FileNotFoundError – If source doesn’t exist

open(file: str | Path, mode: str = 'r', encoding: str | None = None) → TextIO | BinaryIO[source]

Open a file in memory.

Parameters:
  • file – File path to open

  • mode – File mode (r, w, a, rb, wb, etc.)

  • encoding – Text encoding (for text modes, ignored for binary modes)

Returns:

File handle (TextIO for text modes, BinaryIO for binary modes)

Raises:
read_bytes(path: str | Path) → bytes[source]

Read bytes from a file.

Parameters:

path – File path to read

Returns:

File contents as bytes

Raises:

FileNotFoundError – If file doesn’t exist

read_text(path: str | Path, encoding: str = 'utf-8') → str[source]

Read text from a file.

Parameters:
  • path – File path to read

  • encoding – Text encoding (default: utf-8)

Returns:

File contents as string

Raises:

FileNotFoundError – If file doesn’t exist

rmdir(path: str | Path) → None[source]

Remove a directory.

Parameters:

path – Directory path to remove (must be empty)

Raises:

Remove a file.

Parameters:
  • path – File path to remove

  • missing_ok – Don’t raise error if file doesn’t exist

Raises:

FileNotFoundError – If file doesn’t exist and missing_ok is False

write_bytes(path: str | Path, content: bytes) → None[source]

Write bytes to a file.

Parameters:
  • path – File path to write

  • content – Bytes content to write

write_text(path: str | Path, content: str, encoding: str = 'utf-8') → None[source]

Write text to a file.

Parameters:
  • path – File path to write

  • content – Text content to write

  • encoding – Text encoding (default: utf-8)

class pyiv.MockConsole(width: int = 80, height: int = 24)[source]

Bases: BaseConsole

Terminal state machine for testing.

This console maintains a complete terminal state including screen buffer, cursor position, colors, and input state. Use this for testing terminal functionality with full state inspection and event tracking.

Example

>>> from pyiv.console import MockConsole
>>>
>>> console = MockConsole(width=80, height=24)
>>> console.clear()
>>> console.move_cursor(10, 5)
>>> console.set_color(fg=31)  # Red
>>> console.write("Hello")
5
>>>
>>> console.get_cursor()
(15, 5)
>>> console.get_screen_line(5)[10:15]
'Hello'
>>> len(console.get_events())
4
__init__(width: int = 80, height: int = 24)[source]

Initialize terminal state machine.

Parameters:
  • width – Terminal width in columns

  • height – Terminal height in rows

bold(enabled: bool = True) → None[source]

Enable or disable bold text.

clear() → None[source]

Clear the entire screen.

clear_line() → None[source]

Clear the current line.

clear_state() → None[source]

Clear all state (for test setup).

clear_to_end_of_line() → None[source]

Clear from cursor to end of line.

flush() → None[source]

Flush buffer (no-op for mock console).

get_color() → Tuple[int | None, int | None][source]

Get current color state.

Returns:

Tuple of (fg, bg)

get_cursor() → Tuple[int, int][source]

Get current cursor position.

Returns:

Tuple of (x, y)

get_cursor_position() → Tuple[int, int] | None[source]

Get current cursor position.

Returns:

Tuple of (x, y)

get_echo_enabled() → bool[source]

Get echo state.

Returns:

True if echo is enabled

get_events() → List[TerminalEvent][source]

Get event history.

Returns:

List of terminal events

get_raw_mode() → bool[source]

Get raw mode state.

Returns:

True if raw mode is enabled

get_screen() → List[List[str]][source]

Get screen buffer (copy).

Returns:

2D list of screen contents

get_screen_char(x: int, y: int) → str[source]

Get character at position.

Parameters:
  • x – Column (0-based)

  • y – Row (0-based)

Returns:

Character at position

get_screen_line(y: int) → str[source]

Get specific line from screen.

Parameters:

y – Line number (0-based)

Returns:

Line contents as string

get_size() → Tuple[int, int][source]

Get terminal size.

Returns:

Tuple of (columns, rows)

get_style() → Dict[str, bool][source]

Get text style state.

Returns:

Dictionary of style flags

hide_cursor() → None[source]

Hide the cursor.

is_tty() → bool[source]

Check if console is a TTY (always False for mock).

Returns:

False (mock console is not a real TTY)

move_cursor(x: int, y: int) → None[source]

Move cursor to position.

move_cursor_down(n: int = 1) → None[source]

Move cursor down n lines.

move_cursor_home() → None[source]

Move cursor to home position.

move_cursor_left(n: int = 1) → None[source]

Move cursor left n columns.

move_cursor_right(n: int = 1) → None[source]

Move cursor right n columns.

move_cursor_up(n: int = 1) → None[source]

Move cursor up n lines.

read_char(timeout: float | None = None) → str | None[source]

Read a single character from input buffer.

Parameters:

timeout – Optional timeout in seconds (ignored for MockConsole)

Returns:

Character string or None if no input available

read_line(prompt: str = '') → str[source]

Read a line from input buffer.

Parameters:

prompt – Optional prompt string to display

Returns:

Line of text (without newline)

read_password(prompt: str = '') → str[source]

Read password with echo disabled.

reset_color() → None[source]

Reset colors to default.

restore_cursor() → None[source]

Restore saved cursor position.

save_cursor() → None[source]

Save current cursor position.

set_color(fg: int | None = None, bg: int | None = None) → None[source]

Set foreground and/or background color.

set_echo(enabled: bool) → None[source]

Turn echo on/off for password prompts.

set_raw_mode(enabled: bool) → None[source]

Enable/disable raw mode.

show_cursor() → None[source]

Show the cursor.

simulate_input(text: str) → None[source]

Simulate input for testing.

Parameters:

text – Text to add to input buffer

underline(enabled: bool = True) → None[source]

Enable or disable underline.

writable() → bool[source]

Check if console is writable.

Returns:

True (mock console is always writable)

write(s: str) → int[source]

Write string, parsing escape sequences and updating state.

Parameters:

s – String to write

Returns:

Number of characters written

class pyiv.MockDateTimeService(fixed_time: datetime | None = None)[source]

Bases: DateTimeService

Mock datetime service for testing.

Why this exists: Freeze calendar time for deterministic tests of ISO/UTC stamps.

This implementation allows you to control the time returned, making it easy to test time-dependent code with predictable timestamps.

Example

>>> from datetime import datetime, timezone
>>> service = MockDateTimeService(datetime(2024, 1, 15, 10, 30, 0, tzinfo=timezone.utc))
>>> service.now_utc().isoformat()
'2024-01-15T10:30:00+00:00'
>>> service.set_time(datetime(2024, 1, 16, 12, 0, 0, tzinfo=timezone.utc))
>>> service.now_utc().isoformat()
'2024-01-16T12:00:00+00:00'
__init__(fixed_time: datetime | None = None)[source]

Initialize with optional fixed time.

Parameters:

fixed_time – Fixed datetime to return (defaults to current time if not provided)

now_utc() → datetime[source]

Return fixed time.

Returns:

The fixed datetime set for this mock service

now_utc_iso() → str[source]

Return fixed time as ISO string.

Returns:

The fixed datetime as ISO format string

set_time(new_time: datetime) → None[source]

Set the fixed time.

Parameters:

new_time – New fixed time. If no timezone is provided, UTC is assumed.

class pyiv.Multibinder(*args, **kwargs)[source]

Bases: Protocol, Generic[T]

Protocol for binding multiple implementations of the same type.

Why this exists: Register many implementations of one type for injection as a collection.

Multibinders allow multiple implementations of the same type to be registered and injected as a collection (Set or List).

Example

class MyMultibinder(Multibinder[EventHandler]):
def add(self, implementation: Type[EventHandler]) -> None:

# Register implementation pass

__init__(*args, **kwargs)
add(implementation: Type[T]) → None[source]

Add an implementation to the multibinding.

Parameters:

implementation – The implementation class to add

add_instance(instance: T) → None[source]

Add a pre-created instance to the multibinding.

Parameters:

instance – The instance to add

class pyiv.Named(name: str)[source]

Bases: object

String-based qualifier for named bindings.

Why this exists: Most common Qualifier: distinguish bindings with a string name.

This is the most common qualifier type, allowing bindings to be distinguished by a string name.

Example

>>> from pyiv.key import Named
>>>
>>> primary = Named("primary")
>>> replica = Named("replica")
__init__(name: str)[source]

Initialize named qualifier.

Parameters:

name – The name identifier

class pyiv.NetworkClient[source]

Bases: ChainHandler

Abstract network client in the NETWORK_CLIENT chain.

Use this as the injectable type for protocol-specific clients (HTTP, HTTPS, etc.). Register concrete clients in a chain so callers dispatch by URL scheme without hard-coding urllib. Subclasses must implement handler_type and request().

Why this exists: Injectable HTTP-ish client so call sites do not hard-code urllib.

Example

>>> from typing import Any, Dict, Optional, Union
>>> from pyiv.network.base import NetworkClient
>>> class StubHTTP(NetworkClient):
...     @property
...     def handler_type(self) -> str:
...         return "http"
...     def request(
...         self,
...         method: str,
...         url: str,
...         headers: Optional[Dict[str, str]] = None,
...         data: Optional[Union[str, bytes]] = None,
...         timeout: Optional[float] = None,
...     ) -> Dict[str, Any]:
...         return {"status": 200, "headers": {}, "body": b"ok", "url": url}
>>> StubHTTP().request("GET", "http://example.test")["body"]
b'ok'
property chain_type: ChainType

Return the chain type (always NETWORK_CLIENT for NetworkClient).

Returns:

ChainType.NETWORK_CLIENT

handle(request: Any, **kwargs) → Any[source]

Handle a network request.

This is the chain handler interface. For NetworkClient, requests can be: - A tuple of (method, url, headers, data, timeout) -> returns response - A dict with “method”, “url”, etc. keys -> processes accordingly - A direct URL string -> performs GET request

Parameters:
  • request – The request (can be tuple, dict, or URL string)

  • **kwargs – Additional keyword arguments (method, url, headers, data, timeout)

Returns:

Response dictionary with status, headers, and body

abstract property handler_type: str

Return the protocol identifier (e.g., “http”, “https”).

Returns:

A string identifying the network protocol

abstract request(method: str, url: str, headers: Dict[str, str] | None = None, data: str | bytes | None = None, timeout: float | None = None) → Dict[str, Any][source]

Make a network request.

Parameters:
  • method – HTTP method (GET, POST, PUT, DELETE, etc.)

  • url – The URL to request

  • headers – Optional dictionary of HTTP headers

  • data – Optional request body (string or bytes)

  • timeout – Optional timeout in seconds

Returns:

  • status: HTTP status code

  • headers: Response headers dictionary

  • body: Response body (bytes)

  • url: Final URL after redirects

Return type:

Dictionary containing

class pyiv.NoOpSerDe[source]

Bases: SerDe

No-op SerDe that passes through data unchanged.

Use this as the default when no encoding is configured, or when a pipeline already holds strings/bytes and must not re-encode them. Strings and bytes are returned as-is; other objects become str(obj).

Why this exists: Default pass-through encoder when no wire format was configured.

Example

>>> from pyiv.serde.encodings import NoOpSerDe
>>> serde = NoOpSerDe()
>>> serde.handler_type
'noop'
>>> serde.serialize("already-encoded")
'already-encoded'
>>> serde.deserialize("already-encoded")
'already-encoded'
deserialize(data: str | bytes, target_type: Type[T] | None = None) → T[source]

Pass through the data unchanged.

Parameters:
  • data – The encoded data (string or bytes)

  • target_type – Optional type hint (ignored)

Returns:

The data unchanged

property handler_type: str

Return the handler type identifier.

Returns:

The handler type identifier (“noop”)

serialize(obj: Any) → str | bytes[source]

Pass through the object unchanged.

Parameters:

obj – The Python object to serialize

Returns:

The object unchanged (as string or bytes)

class pyiv.NoScope[source]

Bases: object

No scope - creates a new instance every time.

Why this exists: Explicit unscoped binding when a type annotation would otherwise imply reuse.

This scope does not cache instances. Each call to get() will create a new instance. This is the default behavior when no scope is specified.

Example

>>> from pyiv.scope import NoScope
>>> from pyiv import Config
>>>
>>> class MyConfig(Config):
...     def configure(self):
...         # No caching - new instance every time
...         self.register(Logger, FileLogger, scope=NoScope())
scope(key: Type | str | tuple, provider: Provider[Any]) → Provider[Any][source]

Return the provider unchanged (no caching).

Parameters:
  • key – The key identifying the dependency

  • provider – The provider to scope

Returns:

The same provider (no caching)

class pyiv.PTYConsole[source]

Bases: BaseConsole

Pseudoterminal console for testing.

This console creates a pseudoterminal (pty) that behaves like a real terminal. Use this for testing programs that require TTY behavior.

Example

>>> from pyiv.console import PTYConsole
>>>
>>> try:
...     with PTYConsole() as console:
...         assert console.is_tty() is True
...         _ = console.write("Test")
... except (OSError, RuntimeError):
...     pass  # pty may be unavailable in sandbox/CI environments
__init__()[source]

Initialize pseudoterminal console.

bold(enabled: bool = True) → None[source]

Enable or disable bold text.

clear() → None[source]

Clear the entire screen.

clear_line() → None[source]

Clear the current line.

clear_state() → None[source]

Clear all state (no-op for PTYConsole).

clear_to_end_of_line() → None[source]

Clear from cursor to end of line.

close() → None[source]

Close the pty.

flush() → None[source]

Flush pty buffer.

get_color() → Tuple[int | None, int | None][source]

Get color state (not available).

get_cursor() → Tuple[int, int][source]

Get cursor position (not available).

get_cursor_position() → Tuple[int, int] | None[source]

Get cursor position (not easily available from pty).

Returns:

None (not implemented for pty)

get_echo_enabled() → bool[source]

Get echo state (not available).

get_events() → List[TerminalEvent][source]

Get event history (not available).

get_raw_mode() → bool[source]

Get raw mode state (not available).

get_screen() → List[List[str]][source]

Get screen buffer (not available).

get_screen_char(x: int, y: int) → str[source]

Get character at position (not available for PTYConsole).

Parameters:
  • x – Column (0-based)

  • y – Row (0-based)

Returns:

Space character as placeholder

get_screen_line(y: int) → str[source]

Get specific line from screen (not available for PTYConsole).

Parameters:

y – Line number (0-based)

Returns:

Empty string as placeholder

get_size() → Tuple[int, int][source]

Get pty size.

Returns:

Tuple of (columns, rows)

get_style() → Dict[str, bool][source]

Get text style state (not available for PTYConsole).

Returns:

Dictionary with all styles False

hide_cursor() → None[source]

Hide the cursor.

is_tty() → bool[source]

Check if pty is a terminal (always True).

Returns:

True (pty always behaves as TTY)

move_cursor(x: int, y: int) → None[source]

Move cursor to position.

move_cursor_down(n: int = 1) → None[source]

Move cursor down n lines.

move_cursor_home() → None[source]

Move cursor to home position.

move_cursor_left(n: int = 1) → None[source]

Move cursor left n columns.

move_cursor_right(n: int = 1) → None[source]

Move cursor right n columns.

move_cursor_up(n: int = 1) → None[source]

Move cursor up n lines.

read_char(timeout: float | None = None) → str | None[source]

Read a single character.

Parameters:

timeout – Optional timeout in seconds (not fully implemented for PTY)

Returns:

Character string or None if no input available

read_line(prompt: str = '') → str[source]

Read a line of input.

Parameters:

prompt – Optional prompt string to display

Returns:

Line of text (without newline)

read_password(prompt: str = '') → str[source]

Read password (basic implementation).

reset_color() → None[source]

Reset colors to default.

restore_cursor() → None[source]

Restore saved cursor position.

save_cursor() → None[source]

Save current cursor position.

set_color(fg: int | None = None, bg: int | None = None) → None[source]

Set foreground and/or background color.

set_echo(enabled: bool) → None[source]

Turn echo on/off (not implemented for pty).

set_raw_mode(enabled: bool) → None[source]

Enable/disable raw mode (not implemented for pty).

show_cursor() → None[source]

Show the cursor.

underline(enabled: bool = True) → None[source]

Enable or disable underline.

writable() → bool[source]

Check if pty is writable.

Returns:

True (pty is always writable)

write(s: str) → int[source]

Write string to pty.

Parameters:

s – String to write

Returns:

Number of characters written

class pyiv.PickleSerDe[source]

Bases: SerDe

Python pickle SerDe.

Use this when you need to round-trip arbitrary Python objects that JSON, XML, or base64 text cannot represent. Prefer safer text codecs for untrusted input; pickle is a stdlib fallback for trusted local data.

Example

>>> from pyiv.serde.encodings import PickleSerDe
>>> serde = PickleSerDe()
>>> serde.handler_type
'pickle'
>>> serde.deserialize(serde.serialize({"x": 2}))
{'x': 2}
deserialize(data: str | bytes, target_type: Type[T] | None = None) → T[source]

Deserialize using pickle.

Parameters:
  • data – The pickled data (bytes)

  • target_type – Optional type hint (ignored, pickle handles types)

Returns:

Deserialized Python object

property handler_type: str

Return the handler type identifier.

Returns:

The handler type identifier (“pickle”)

serialize(obj: Any) → bytes[source]

Serialize using pickle.

Parameters:

obj – The Python object to serialize

Returns:

Pickled bytes representation

class pyiv.PrivateConfig[source]

Bases: Config

Config whose bindings are hidden unless explicitly exposed.

Why this exists: Hide internal bindings and expose only a facade type to the parent graph.

When installed into a parent config, a child injector owns the private graph. Only types/keys passed to expose() are visible to the parent (as providers that delegate into the child).

Example

>>> from pyiv import Config, PrivateConfig, get_injector
>>> class Hidden:
...     pass
>>> class Service:
...     def __init__(self, hidden: Hidden):
...         self.hidden = hidden
>>> class Impl(PrivateConfig):
...     def configure(self):
...         self.register(Hidden, Hidden)
...         self.register(Service, Service)
...         self.expose(Service)
>>> class App(Config):
...     def configure(self):
...         self.install(Impl)
>>> svc = get_injector(App).inject(Service)
>>> isinstance(svc.hidden, Hidden)
True
expose(type_or_key: Type | Key[Any]) → None[source]

Expose a binding to the parent environment when this module is installed.

class pyiv.Provider(*args, **kwargs)[source]

Bases: Protocol, Generic[T]

Protocol for provider implementations.

Why this exists: Lazy or repeated creation of a type without constructing it at inject() time; inject Provider[T] instead of T.

Providers are used to supply instances of a specific type, allowing for customized creation logic, lazy initialization, and injector access.

This is the standard pattern in dependency injection frameworks for providing instances. Providers can be injected themselves, allowing for lazy initialization and multiple instance creation.

Example

>>> from pyiv.provider import Provider
>>> from pyiv import Config, Injector, get_injector
>>>
>>> class Database:
...     pass
>>>
>>> class DatabaseProvider:
...     def __init__(self, injector: Injector):
...         self._injector = injector
...
...     def get(self) -> Database:
...         return self._injector.inject(Database)
>>>
>>> class MyConfig(Config):
...     def configure(self):
...         self.register(Database, Database)
>>>
>>> injector = get_injector(MyConfig)
>>> provider: Provider[Database] = DatabaseProvider(injector)
>>> isinstance(provider.get(), Database)
True
__init__(*args, **kwargs)
get() → T[source]

Get an instance of type T.

Returns:

An instance of type T

class pyiv.PythonDateTimeService[source]

Bases: DateTimeService

DateTime service backed by the system clock.

Use this in production for real UTC timestamps. In tests, swap in MockDateTimeService so assertions do not depend on wall-clock time.

Why this exists: Production DateTimeService backed by the system clock.

Example

>>> from datetime import datetime, timezone
>>> from pyiv.datetime_service import PythonDateTimeService
>>> svc = PythonDateTimeService()
>>> now = svc.now_utc()
>>> isinstance(now, datetime) and now.tzinfo == timezone.utc
True
>>> "T" in svc.now_utc_iso()
True
now_utc() → datetime[source]

Get current UTC datetime from system clock using Python’s datetime.

Returns:

Current datetime in UTC timezone

now_utc_iso() → str[source]

Get current UTC datetime as ISO format string using Python’s datetime.

Returns:

Current datetime in UTC as ISO format string

class pyiv.Qualifier(*args, **kwargs)[source]

Bases: Protocol

Marker for distinguishing multiple bindings of the same type.

Why this exists: One interface often has several implementations (primary vs replica DB). A qualifier + Key selects which binding to inject without inventing wrapper types.

Example

>>> from pyiv.key import Named
>>> isinstance(Named("primary"), Named)
True
__init__(*args, **kwargs)
class pyiv.RealClock[source]

Bases: Clock

Real clock implementation using standard library.

Why this exists: Production binding for real wall-clock time and sleeping timers.

This is the production implementation that uses Python’s built-in time and threading modules to provide actual system clock time and real sleep operations.

Example

>>> clock = RealClock()
>>> isinstance(clock.time(), float)
True
>>> clock.monotonic() > 0
True
monotonic() → float[source]

Get monotonic time (not affected by system clock adjustments).

Returns:

Monotonic time as float (seconds since an arbitrary point)

sleep(seconds: float) → None[source]

Sleep for specified duration.

Parameters:

seconds – Duration to sleep in seconds

start_timer(interval: float, callback: Callable[[], None], repeat: bool = False) → Timer[source]

Start a timer using threading.Timer.

Parameters:
  • interval – Time interval in seconds before callback is called

  • callback – Function to call when timer fires

  • repeat – If True, timer will repeat after each interval

Returns:

Timer object that can be used to cancel the timer

thread_sleep(seconds: float) → None[source]

Sleep in current thread.

Parameters:

seconds – Duration to sleep in seconds

time() → float[source]

Get current time as seconds since epoch.

Returns:

Current time as float (seconds since Unix epoch)

class pyiv.RealConsole(stream: TextIO | None = None, stdin: TextIO | None = None)[source]

Bases: BaseConsole

Production console implementation using sys.stdout.

This console wraps sys.stdout, providing the standard console output behavior and TTY-specific functionality. Use this in production code for normal console output and terminal features like spinners, animations, and password prompts.

Why this exists: Production Console that writes to a real stream/TTY.

Example

>>> from io import StringIO
>>> from pyiv.console import RealConsole
>>>
>>> buf = StringIO()
>>> console = RealConsole(stream=buf)
>>> print("Hello, World!", file=console)
>>> "Hello, World!" in buf.getvalue()
True
__init__(stream: TextIO | None = None, stdin: TextIO | None = None)[source]

Initialize console with streams.

Parameters:
  • stream – TextIO stream to use for output (defaults to sys.stdout)

  • stdin – TextIO stream to use for input (defaults to sys.stdin)

bold(enabled: bool = True) → None[source]

Enable or disable bold text.

Parameters:

enabled – True to enable bold, False to disable

clear() → None[source]

Clear the entire screen.

clear_line() → None[source]

Clear the current line.

clear_state() → None[source]

Clear all state (no-op for RealConsole).

clear_to_end_of_line() → None[source]

Clear from cursor to end of line.

flush() → None[source]

Flush stdout buffer.

get_color() → Tuple[int | None, int | None][source]

Get current color state (not available for RealConsole).

Returns:

Tuple of (None, None) as placeholder

get_cursor() → Tuple[int, int][source]

Get current cursor position (not available for RealConsole).

Returns:

Tuple of (0, 0) as placeholder

get_cursor_position() → Tuple[int, int] | None[source]

Get current cursor position.

Note: This requires reading a response from the terminal, which may not be available in all contexts.

Returns:

Tuple of (x, y) or None if not available

get_echo_enabled() → bool[source]

Get echo state (not available for RealConsole).

Returns:

True as placeholder

get_events() → List[TerminalEvent][source]

Get event history (not available for RealConsole).

Returns:

Empty list as placeholder

get_raw_mode() → bool[source]

Get raw mode state (not available for RealConsole).

Returns:

False as placeholder

get_screen() → List[List[str]][source]

Get screen buffer (not available for RealConsole).

Returns:

Empty list as placeholder

get_screen_char(x: int, y: int) → str[source]

Get character at position (not available for RealConsole).

Parameters:
  • x – Column (0-based)

  • y – Row (0-based)

Returns:

Space character as placeholder

get_screen_line(y: int) → str[source]

Get specific line from screen (not available for RealConsole).

Parameters:

y – Line number (0-based)

Returns:

Empty string as placeholder

get_size() → Tuple[int, int][source]

Get terminal size.

Returns:

Tuple of (columns, rows)

get_style() → Dict[str, bool][source]

Get text style state (not available for RealConsole).

Returns:

Dictionary with all styles False

hide_cursor() → None[source]

Hide the cursor.

is_tty() → bool[source]

Check if stdout is a real terminal.

Returns:

True if stdout is connected to a TTY

move_cursor(x: int, y: int) → None[source]

Move cursor to position.

Parameters:
  • x – Column (0-based, but ANSI uses 1-based)

  • y – Row (0-based, but ANSI uses 1-based)

move_cursor_down(n: int = 1) → None[source]

Move cursor down n lines.

move_cursor_home() → None[source]

Move cursor to home position (0, 0).

move_cursor_left(n: int = 1) → None[source]

Move cursor left n columns.

move_cursor_right(n: int = 1) → None[source]

Move cursor right n columns.

move_cursor_up(n: int = 1) → None[source]

Move cursor up n lines.

read_char(timeout: float | None = None) → str | None[source]

Read a single character (requires raw mode).

Parameters:

timeout – Optional timeout in seconds

Returns:

Character string or None if no input available

read_line(prompt: str = '') → str[source]

Read a line of input.

Parameters:

prompt – Optional prompt string to display

Returns:

Line of text (without newline)

read_password(prompt: str = '') → str[source]

Read password with echo disabled.

Parameters:

prompt – Optional prompt to display

Returns:

Password string

reset_color() → None[source]

Reset colors to default.

restore_cursor() → None[source]

Restore saved cursor position.

save_cursor() → None[source]

Save current cursor position.

set_color(fg: int | None = None, bg: int | None = None) → None[source]

Set foreground and/or background color.

Parameters:
  • fg – Foreground color code (30-37 for basic, 90-97 for bright)

  • bg – Background color code (40-47 for basic, 100-107 for bright)

set_echo(enabled: bool) → None[source]

Turn echo on/off for password prompts.

Parameters:

enabled – True to enable echo, False to disable

set_raw_mode(enabled: bool) → None[source]

Enable/disable raw mode for character-by-character input.

Parameters:

enabled – True for raw mode, False for cooked mode

show_cursor() → None[source]

Show the cursor.

underline(enabled: bool = True) → None[source]

Enable or disable underline.

Parameters:

enabled – True to enable underline, False to disable

writable() → bool[source]

Check if stdout is writable.

Returns:

True if stdout is writable

write(s: str) → int[source]

Write string to stdout.

Parameters:

s – String to write

Returns:

Number of characters written

class pyiv.RealFilesystem[source]

Bases: Filesystem

Real filesystem implementation using the standard library.

Use this in production when you need actual disk I/O. Prefer MemoryFilesystem in unit tests so examples and CI never write into the working tree.

Why this exists: Production Filesystem that performs real disk I/O.

Example

>>> import tempfile
>>> from pathlib import Path
>>> from pyiv.filesystem import RealFilesystem
>>> fs = RealFilesystem()
>>> with tempfile.TemporaryDirectory() as d:
...     p = Path(d) / "hello.txt"
...     fs.write_text(p, "hi")
...     fs.read_text(p)
'hi'
copy(src: str | Path, dst: str | Path) → None[source]

Copy a file.

Parameters:
  • src – Source file path

  • dst – Destination file path

exists(path: str | Path) → bool[source]

Check if path exists.

Parameters:

path – Path to check

Returns:

True if path exists, False otherwise

get_size(path: str | Path) → int[source]

Get file size in bytes.

Parameters:

path – File path

Returns:

File size in bytes

glob(pattern: str | Path) → Iterator[Path][source]

Glob pattern matching.

Parameters:

pattern – Glob pattern to match

Yields:

Matching paths

is_dir(path: str | Path) → bool[source]

Check if path is a directory.

Parameters:

path – Path to check

Returns:

True if path is a directory, False otherwise

is_file(path: str | Path) → bool[source]

Check if path is a file.

Parameters:

path – Path to check

Returns:

True if path is a file, False otherwise

listdir(path: str | Path) → Iterator[str][source]

List directory contents.

Parameters:

path – Directory path to list

Yields:

Directory entry names

mkdir(path: str | Path, parents: bool = False, exist_ok: bool = False) → None[source]

Create a directory.

Parameters:
  • path – Directory path to create

  • parents – Create parent directories if needed

  • exist_ok – Don’t raise error if directory already exists

move(src: str | Path, dst: str | Path) → None[source]

Move/rename a file or directory.

Parameters:
  • src – Source path

  • dst – Destination path

open(file: str | Path, mode: str = 'r', encoding: str | None = None) → TextIO | BinaryIO[source]

Open a file using built-in open().

Parameters:
  • file – File path to open

  • mode – File mode (r, w, a, rb, wb, etc.)

  • encoding – Text encoding (for text modes, ignored for binary modes)

Returns:

File handle (TextIO for text modes, BinaryIO for binary modes)

read_bytes(path: str | Path) → bytes[source]

Read bytes from a file.

Parameters:

path – File path to read

Returns:

File contents as bytes

read_text(path: str | Path, encoding: str = 'utf-8') → str[source]

Read text from a file.

Parameters:
  • path – File path to read

  • encoding – Text encoding (default: utf-8)

Returns:

File contents as string

rmdir(path: str | Path) → None[source]

Remove a directory.

Parameters:

path – Directory path to remove (must be empty)

Remove a file.

Parameters:
  • path – File path to remove

  • missing_ok – Don’t raise error if file doesn’t exist

write_bytes(path: str | Path, content: bytes) → None[source]

Write bytes to a file.

Parameters:
  • path – File path to write

  • content – Bytes content to write

write_text(path: str | Path, content: str, encoding: str = 'utf-8') → None[source]

Write text to a file.

Parameters:
  • path – File path to write

  • content – Text content to write

  • encoding – Text encoding (default: utf-8)

class pyiv.ReflectionConfig[source]

Bases: Config

Configuration class that supports module-based discovery of implementations.

Why this exists: Auto-discover interface implementations in a package instead of listing every class.

This class extends Config to add reflection-based discovery capabilities. Instead of manually registering each implementation, you can register a package to scan, and all implementations of an interface will be automatically discovered and registered.

Example:

class MyConfig(ReflectionConfig):
    def configure(self):
        self.register_module(
            Handler,
            "my_service.handlers",
            pattern="*Handler",
            singleton_type=SingletonType.SINGLETON,
        )
__init__()[source]

Initialize the reflection configuration.

discover_implementations(interface: Type) → Dict[str, Type][source]

Discover all implementations of an interface in registered modules.

This method scans the registered package for classes that implement the given interface. Only classes within the specified package (and optionally submodules) are discovered - no discovery happens outside the registered package boundaries.

Parameters:

interface – The interface to discover implementations for

Returns:

Dictionary mapping implementation names to classes. Keys are class names (or “submodule.ClassName” for submodules).

Raises:
  • ValueError – If no module registration exists for the interface

  • ImportError – If the package cannot be imported

register_module(interface: Type, package_path: str, pattern: str | None = None, recursive: bool = True, singleton_type: SingletonType = SingletonType.SINGLETON)[source]

Register a package to scan for interface implementations.

Parameters:
  • interface – The abstract class or interface to find implementations of

  • package_path – Python package path (e.g., “my_service.mcp.handlers”)

  • pattern – Optional name pattern for filtering (e.g. *Handler, handle_*). Uses fnmatch syntax. If None, all implementations are discovered.

  • recursive – Whether to scan submodules recursively (default: True)

  • singleton_type – How to handle instances (default: SINGLETON for per-injector reuse)

Raises:

Example:

self.register_module(
    Handler,
    "my_service.handlers",
    pattern="*Handler",
)

self.register_module(
    Service,
    "my_service.services",
    recursive=True,
)
class pyiv.Scope(*args, **kwargs)[source]

Bases: Protocol

Protocol for scope implementations.

Why this exists: Control instance lifetime beyond one-shot construction (request, singleton, custom).

Scopes control the lifecycle of instances created by the dependency injection system. A scope can cache instances, create new ones on demand, or implement custom lifecycle logic.

Scopes are more flexible than simple singletons - they can implement request-scoped, thread-scoped, session-scoped, or any custom lifecycle.

Example:

class RequestScope(Scope):
    def __init__(self):
        self._request_cache = {}

    def scope(self, key, provider):
        request_id = get_current_request_id()
        cache_key = (key, request_id)
        if cache_key not in self._request_cache:
            self._request_cache[cache_key] = provider.get()
        return lambda: self._request_cache[cache_key]
__init__(*args, **kwargs)
scope(key: Type | str | tuple, provider: Provider[Any]) → Provider[Any][source]

Scope a provider to this scope’s lifecycle.

Parameters:
  • key – The key identifying the dependency (Type, str, or tuple)

  • provider – The provider to scope

Returns:

A new provider that respects this scope’s lifecycle

class pyiv.SerDe[source]

Bases: ChainHandler

Abstract base class for serialization/deserialization operations.

SerDe is a chain handler for the ENCODING chain. Subclass it for each codec (JSON, base64, XML, pickle, etc.); YAML lives in pyiv-common.

Use this when callers should pick a codec by handler_type instead of hard-coding json.dumps / pickle.loads. Register implementations on the ENCODING chain so injectors and pipelines can swap formats without rewriting call sites. Multiple handlers may share a handler_type with different behaviors (date formatting, null handling, and so on).

Subclasses must implement handler_type, serialize, and deserialize.

Example

>>> import json
>>> from typing import Any, Optional, Type
>>> from pyiv.serde.base import SerDe
>>> class MyJSONSerDe(SerDe):
...     @property
...     def handler_type(self) -> str:
...         return "json"
...     def serialize(self, obj: Any) -> str:
...         return json.dumps(obj)
...     def deserialize(
...         self, data: str, target_type: Optional[Type] = None
...     ) -> Any:
...         return json.loads(data)
>>> serde = MyJSONSerDe()
>>> serde.deserialize(serde.serialize({"a": 1}))
{'a': 1}
property chain_type: ChainType

Return the chain type (always ENCODING for SerDe).

Returns:

ChainType.ENCODING

abstract deserialize(data: str | bytes, target_type: Type[T] | None = None) → T[source]

Deserialize encoded data back to a Python object.

Parameters:
  • data – The encoded data (string or bytes)

  • target_type – Optional type hint for the expected result type

Returns:

Deserialized Python object

handle(request: Any, **kwargs) → Any[source]

Handle a serialization/deserialization request.

This is the chain handler interface. For SerDe, requests can be: - A tuple of (“serialize”, obj) -> returns serialized data - A tuple of (“deserialize”, data, target_type) -> returns deserialized object - A dict with “action” key -> processes accordingly

Parameters:
  • request – The request (can be tuple, dict, or direct object)

  • **kwargs – Additional keyword arguments

Returns:

The result of the operation

abstract property handler_type: str

Return the encoding type identifier (e.g., “json”, “base64”, “pickle”).

Returns:

A string identifying the encoding format

abstract serialize(obj: Any) → str | bytes[source]

Serialize a Python object to encoded format.

Parameters:

obj – The Python object to serialize

Returns:

Serialized representation as string or bytes

class pyiv.ServiceCommand(args: Namespace, injector: Any | None = None)[source]

Bases: Command

Base class for long-running service commands.

Use this when a command should run until interrupted (daemons, workers). execute() runs init() → run() → cleanup() and maps KeyboardInterrupt to exit code 130. Override those hooks; do not reimplement the lifecycle unless necessary.

Example

>>> import argparse
>>> from pyiv.command import ServiceCommand
>>> class OnceService(ServiceCommand):
...     @classmethod
...     def get_name(cls) -> str:
...         return "once"
...     def run(self) -> None:
...         pass  # exit immediately after init
>>> OnceService(argparse.Namespace()).execute()
0
execute() → int[source]

Execute the service command lifecycle.

Returns:

Exit code (0 for success, non-zero for error)

class pyiv.SetMultibinder(interface: Type[T], config: Any)[source]

Bases: Generic[T]

Multibinder that binds to a Set[T].

Why this exists: Collect unique implementations as Set[T] (order not preserved).

This multibinder collects implementations into a set, ensuring uniqueness. Order is not preserved.

Example

>>> from pyiv import Config
>>> class EventHandler:
...     pass
>>> class EmailEventHandler(EventHandler):
...     pass
>>> class MyConfig(Config):
...     def configure(self):
...         pass
>>> multibinder = SetMultibinder(EventHandler, MyConfig())
>>> multibinder.add(EmailEventHandler)
>>> EmailEventHandler in multibinder.get_implementations()
True
__init__(interface: Type[T], config: Any)[source]

Initialize set multibinder.

Parameters:
  • interface – The interface type

  • config – The config to register bindings with

add(implementation: Type[T]) → None[source]

Add an implementation class.

Parameters:

implementation – The implementation class to add

add_instance(instance: T) → None[source]

Add a pre-created instance.

Parameters:

instance – The instance to add

get_implementations() → Set[Type[T]][source]

Get all registered implementation classes.

Returns:

Set of implementation classes

get_instances() → Set[T][source]

Get all registered instances.

Returns:

Set of instances

class pyiv.SimpleFactory(callable_factory: Callable[[...], T])[source]

Bases: Generic[T]

Simple factory that wraps a callable.

Why this exists: Wrap a function/constructor as a Factory without defining a subclass.

Useful for creating factories from functions or constructors without needing to define a full class.

Example

>>> class User:
...     def __init__(self, name: str):
...         self.name = name
>>> def create_user(name: str) -> User:
...     return User(name=name)
>>> from pyiv.factory import SimpleFactory
>>> factory = SimpleFactory(create_user)
>>> factory.create("Alice").name
'Alice'
__init__(callable_factory: Callable[[...], T])[source]

Initialize factory with a callable.

Parameters:

callable_factory – A callable (function, class, etc.) that creates instances

create(*args: Any, **kwargs: Any) → T[source]

Create an instance using the wrapped callable.

Parameters:
  • *args – Positional arguments passed to the callable

  • **kwargs – Keyword arguments passed to the callable

Returns:

An instance created by the callable

class pyiv.SingletonScope[source]

Bases: object

Per-injector singleton scope.

Why this exists: One instance per injector — default app-wide singleton without process globals.

This scope caches instances per injector. Each injector will have its own singleton instance. This is useful when you want singletons but need different instances for different injectors (e.g., test vs production).

Example

>>> from pyiv.scope import SingletonScope
>>> from pyiv import Config
>>>
>>> class MyConfig(Config):
...     def configure(self):
...         # Per-injector singleton
...         self.register(Logger, FileLogger, scope=SingletonScope())
__init__()[source]

Initialize the singleton scope.

scope(key: Type | str | tuple, provider: Provider[Any]) → Provider[Any][source]

Scope provider to per-injector singleton.

Parameters:
  • key – The key identifying the dependency

  • provider – The provider to scope

Returns:

A provider that returns the same instance for this scope

class pyiv.SingletonType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Legacy enum for singleton lifecycle on Config.register.

Why this exists: Older call sites use singleton_type=... instead of a Scope. Prefer Scope for new code; this enum remains for compatibility.

NONE

No singleton behavior - new instance created each time

SINGLETON

Per-injector singleton - one instance per Injector instance

GLOBAL_SINGLETON

Global singleton - one instance shared across all injectors (thread-safe)

Example

>>> from pyiv import Config, get_injector
>>> class Logger:
...     pass
>>> class MyConfig(Config):
...     def configure(self):
...         self.register(Logger, Logger, singleton_type=SingletonType.SINGLETON)
>>> inj = get_injector(MyConfig)
>>> inj.inject(Logger) is inj.inject(Logger)
True
GLOBAL_SINGLETON = 'global_singleton'
NONE = 'none'
SINGLETON = 'singleton'
class pyiv.Stage(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Injector construction stage: lazy vs fail-fast singletons.

Why this exists: Lazy singletons hide binding mistakes until first use. Pass stage=Stage.PRODUCTION to get_injector() so singleton-scoped bindings are created at boot.

DEVELOPMENT

Singletons are created lazily on first inject (default).

PRODUCTION

Singleton-scoped bindings are created when the injector is built.

Example

>>> from pyiv import Config, Stage, get_injector
>>> from pyiv.scope import SingletonScope
>>> class Database:
...     pass
>>> class MyConfig(Config):
...     def configure(self):
...         self.get_binder().bind(Database).to(Database).in_scope(SingletonScope())
>>> isinstance(get_injector(MyConfig, stage=Stage.PRODUCTION).inject(Database), Database)
True
DEVELOPMENT = 'development'
PRODUCTION = 'production'
class pyiv.SyntheticClock(start_time: float = 0.0)[source]

Bases: Clock

Synthetic clock for testing - allows manual time control.

This implementation allows you to control time manually, making it easy to test time-dependent code with predictable timestamps. Time advances only when you call advance() or set_time().

Why this exists: Controllable clock for tests — advance time without sleeping.

Example

>>> clock = SyntheticClock(start_time=100.0)
>>> assert clock.time() == 100.0
>>> clock.advance(5.0)
>>> assert clock.time() == 105.0
>>> clock.set_time(200.0)
>>> assert clock.time() == 200.0
__init__(start_time: float = 0.0)[source]

Initialize synthetic clock.

Parameters:

start_time – Initial time value

advance(seconds: float) → None[source]

Manually advance time.

Parameters:

seconds – Amount to advance time by

monotonic() → float[source]

Get monotonic time.

Returns:

Monotonic time value (same as time() for synthetic clock)

set_time(time_value: float) → None[source]

Set the current time.

Parameters:

time_value – Time value to set

sleep(seconds: float) → None[source]

Advance time instead of actually sleeping.

Parameters:

seconds – Amount of time to advance (does not actually sleep)

start_timer(interval: float, callback: Callable[[], None], repeat: bool = False) → SyntheticTimer[source]

Start a synthetic timer.

Parameters:
  • interval – Time interval in seconds before callback is called

  • callback – Function to call when timer fires

  • repeat – If True, timer will repeat after each interval

Returns:

SyntheticTimer object that can be used to cancel the timer

thread_sleep(seconds: float) → None[source]

Advance time instead of actually sleeping.

Parameters:

seconds – Amount of time to advance (does not actually sleep)

time() → float[source]

Get current synthetic time.

Returns:

Current synthetic time value

class pyiv.Timer[source]

Bases: ABC

Abstract handle for a scheduled callback.

Use timers when code needs one-shot or repeating callbacks driven by a Clock. Prefer SyntheticClock.start_timer in tests so time advances only when you call advance(), not wall-clock sleep.

Example

>>> from pyiv.clock import SyntheticClock
>>> clock = SyntheticClock(start_time=0.0)
>>> fired = []
>>> timer = clock.start_timer(5.0, lambda: fired.append(True), repeat=False)
>>> timer.is_active()
True
>>> clock.advance(5.0)
>>> fired
[True]
>>> timer.cancel()
abstract cancel() → None[source]

Cancel the timer.

Returns:

None

abstract is_active() → bool[source]

Check if timer is still active.

Returns:

True if active, False otherwise

class pyiv.XMLSerDe[source]

Bases: SerDe

XML encoding SerDe for simple dict/list structures.

Use this when interoperability with XML-oriented systems matters and the payload is a shallow dict or list. Nested dicts become child elements; lists become item children. Not a full schema or namespace mapper.

Example

>>> from pyiv.serde.encodings import XMLSerDe
>>> serde = XMLSerDe()
>>> serde.handler_type
'xml'
>>> xml = serde.serialize({"name": "Ada"})
>>> xml
'<root><name>Ada</name></root>'
>>> serde.deserialize(xml)
{'name': 'Ada'}
deserialize(data: str | bytes, target_type: Type[T] | None = None) → T[source]

Deserialize XML string/bytes back to a Python object.

Parameters:
  • data – The XML string or bytes

  • target_type – Optional type hint (returns dict by default)

Returns:

Deserialized Python object (dict or list)

property handler_type: str

Return the handler type identifier.

Returns:

The handler type identifier (“xml”)

serialize(obj: Any) → str[source]

Serialize using XML encoding.

Parameters:

obj – The Python object to serialize (dict or list)

Returns:

XML string representation

pyiv.get_injector(config: Type[Config] | Config, *, stage: Stage = Stage.DEVELOPMENT) → Injector[source]

Create an injector from a configuration class or instance.

Parameters:
  • config – A Config subclass or Config instance that defines dependencies

  • stage – DEVELOPMENT (default, lazy singletons) or PRODUCTION (eager)

Returns:

An Injector instance configured with the given config

Example

>>> from pyiv import Config, get_injector
>>> class Database:
...     pass
>>> class PostgreSQL(Database):
...     pass
>>> class MyConfig(Config):
...     def configure(self):
...         self.register(Database, PostgreSQL)
>>> isinstance(get_injector(MyConfig).inject(Database), PostgreSQL)
True
pyiv.get_optional_type(annotation: Any) → Type | None[source]

Extract the inner type from Optional[T].

Parameters:

annotation – The Optional[T] annotation

Returns:

The inner type T, or None if not Optional

pyiv.is_optional_type(annotation: Any) → bool[source]

Check if a type annotation is Optional[T].

Parameters:

annotation – The type annotation to check

Returns:

True if the annotation is Optional[T], False otherwise

pyiv.override(*bases: Type[Config] | Config) → OverrideBuilder[source]

Start an override overlay over one or more base configs.

Example:

get_injector(override(ProdConfig).with_(TestConfig))