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:
SerDeBase64 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
- class pyiv.BaseConsole[source]
Bases:
ABCAbstract base class for console implementations.
Subclass this when you need a custom
Console(file-like write plus optional TTY helpers). Implementwrite/flush/writable; terminal methods default to no-ops so simple sinks stay small. PreferMemoryConsolefor capturingprint()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'
- get_events() List[TerminalEvent][source]
Get event history (default: empty list).
- 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_style() Dict[str, bool][source]
Get text style state (default: all False).
- Returns:
Dictionary of style flags
- read_char(timeout: float | None = None) str | None[source]
Read a single character (default: None).
- read_password(prompt: str = '') str[source]
Read password with echo disabled (default: empty string).
- set_color(fg: int | None = None, bg: int | None = None) None[source]
Set foreground and/or background color (no-op by default).
- class pyiv.BaseFactory[source]
-
ABC when you want a class-based factory with constructor injection.
Why this exists: Protocols are fine for typing; subclassing
BaseFactorygives a real type the injector can construct, with dependencies supplied to__init__and runtime args tocreate.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'
- class pyiv.BaseProvider[source]
-
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'
- class pyiv.Binder(*args, **kwargs)[source]
Bases:
ProtocolFluent configuration API for contributing bindings to a
Config.Why this exists: Direct
Config.register*calls work, but complex graphs read better asbind(X).to(Y).in_scope(Z). The binder also supportsinstall,expose, andrequire_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.
- class pyiv.BindingBuilder(*args, **kwargs)[source]
-
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 sameConfigstore.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:
CommandBase class for one-shot CLI commands (not long-running services).
Use this for tools that init, run once, and exit. Override
run()(and optionallyinit()/cleanup()); setself._exit_codefor non-zero status. Do not overrideexecute()— 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.
- class pyiv.ChainHandler[source]
Bases:
ABCPluggable 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
Configand resolve them withinject_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
- class pyiv.ChainType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]
Bases:
EnumCategory 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.
ChainTypenamespaces 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:
ABCAbstract 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
- class pyiv.Command(args: Namespace, injector: Any | None = None)[source]
Bases:
ABCBase 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
CLICommandfor one-shot tools andServiceCommandfor 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.
- class pyiv.CommandRunner(config: Any | None = None)[source]
Bases:
objectDiscovers and runs
Commandsubclasses 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:
objectModule 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, overrideconfigure(), and register interfaces → implementations (or useget_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
- 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;
PrivateConfiguses the set when installed into a parent.
- 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).
PrivateConfiginstances 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
otherinto this config.- Parameters:
other – Source config
replace – If True,
otheroverwrites 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:
TypeError – If instance is not a ChainHandler
ValueError – If name is empty
- 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
- class pyiv.Console(*args, **kwargs)[source]
Bases:
ProtocolProtocol 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)
- clear_state() None[source]
Clear all state (for test setup).
Resets terminal state including screen buffer, cursor position, colors, styles, and event history.
- 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.)
- 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)
- 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
- 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
- exception pyiv.CreationError(message: str, *, path: Sequence[str] | None = None, causes: Sequence[BaseException] | None = None)[source]
Bases:
ExceptionRaised 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 -> Databaseinstead of a bareValueError.- 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
- class pyiv.DateTimeService[source]
Bases:
ABCAbstract interface for UTC datetime operations.
Depend on this instead of calling
datetime.nowdirectly so production code can usePythonDateTimeServicewhile tests injectMockDateTimeServicewith 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'
- class pyiv.Factory(*args, **kwargs)[source]
-
Callable creation API for objects that need runtime arguments.
Why this exists: A DI
Providerbuilds one graph-managed instance. AFactorycreates 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)
- 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
- class pyiv.FileConsole(file: str | Path, mode: str = 'w', encoding: str = 'utf-8')[source]
Bases:
BaseConsoleFile-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)
- class pyiv.Filesystem[source]
Bases:
ABCAbstract filesystem for injectable file I/O.
Why this exists: Production code that calls
open()/pathlibdirectly is hard to unit-test without touching disk or mocking builtins. Depend onFilesystemand bindRealFilesystemin production andMemoryFilesystemin 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
- abstract unlink(path: str | Path, missing_ok: bool = False) None[source]
Remove a file.
- Parameters:
path – File path
missing_ok – Don’t raise error if file doesn’t exist
- class pyiv.GlobalSingletonRegistry[source]
Bases:
objectProcess-wide store for
GLOBAL_SINGLETONinstances.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
- class pyiv.GlobalSingletonScope[source]
Bases:
objectGlobal 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)
- class pyiv.HTTPClient[source]
Bases:
NetworkClientHTTP(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:
NetworkClientHTTPS-only client using stdlib
urllib.Use this when you want to reject non-HTTPS URLs at the client boundary (scheme must be
https://). LikeHTTPClient, 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:
ValueError – If URL does not start with https://
URLError – If the request fails
- class pyiv.Injector(config: Config, *, stage: Stage = Stage.DEVELOPMENT, parent: Injector | None = None, wire_private: bool = True)[source]
Bases:
objectResolves types and keys from a
Configgraph.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.
- 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
- 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'
- class pyiv.JSONSerDe[source]
Bases:
SerDeStandard JSON SerDe using Python’s
jsonmodule.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
- 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
- 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
- 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
- class pyiv.MembersInjector(*args, **kwargs)[source]
-
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)
- class pyiv.MemoryConsole[source]
Bases:
BaseConsoleIn-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'
- close() None[source]
Close the buffer.
Note: After closing, getvalue() will raise ValueError. To get the value before closing, call getvalue() first.
- 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
- class pyiv.MemoryFilesystem[source]
Bases:
FilesystemIn-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'
- 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:
FileExistsError – If directory or file exists at path and exist_ok is False
FileNotFoundError – If parent directory doesn’t exist and parents is False
- 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:
FileNotFoundError – If file doesn’t exist in read mode
ValueError – If mode is unsupported
- 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:
FileNotFoundError – If directory doesn’t exist
ValueError – If trying to remove root directory
OSError – If directory is not empty
- unlink(path: str | Path, missing_ok: bool = False) None[source]
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
- class pyiv.MockConsole(width: int = 80, height: int = 24)[source]
Bases:
BaseConsoleTerminal 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
- get_color() Tuple[int | None, int | None][source]
Get current color state.
- Returns:
Tuple of (fg, bg)
- get_cursor_position() Tuple[int, int] | None[source]
Get current cursor position.
- Returns:
Tuple of (x, y)
- get_events() List[TerminalEvent][source]
Get event history.
- Returns:
List of terminal events
- 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
- is_tty() bool[source]
Check if console is a TTY (always False for mock).
- Returns:
False (mock console is not a real TTY)
- 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)
- set_color(fg: int | None = None, bg: int | None = None) None[source]
Set foreground and/or background color.
- simulate_input(text: str) None[source]
Simulate input for testing.
- Parameters:
text – Text to add to input buffer
- class pyiv.MockDateTimeService(fixed_time: datetime | None = None)[source]
Bases:
DateTimeServiceMock 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
- class pyiv.Multibinder(*args, **kwargs)[source]
-
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)
- class pyiv.Named(name: str)[source]
Bases:
objectString-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")
- class pyiv.NetworkClient[source]
Bases:
ChainHandlerAbstract 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_typeandrequest().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:
SerDeNo-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
- class pyiv.NoScope[source]
Bases:
objectNo 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())
- class pyiv.PTYConsole[source]
Bases:
BaseConsolePseudoterminal 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
- get_cursor_position() Tuple[int, int] | None[source]
Get cursor position (not easily available from pty).
- Returns:
None (not implemented for pty)
- get_events() List[TerminalEvent][source]
Get event history (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_style() Dict[str, bool][source]
Get text style state (not available for PTYConsole).
- Returns:
Dictionary with all styles False
- is_tty() bool[source]
Check if pty is a terminal (always True).
- Returns:
True (pty always behaves as TTY)
- 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)
- class pyiv.PickleSerDe[source]
Bases:
SerDePython 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
- class pyiv.PrivateConfig[source]
Bases:
ConfigConfig 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
- class pyiv.Provider(*args, **kwargs)[source]
-
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)
- class pyiv.PythonDateTimeService[source]
Bases:
DateTimeServiceDateTime service backed by the system clock.
Use this in production for real UTC timestamps. In tests, swap in
MockDateTimeServiceso 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
- class pyiv.Qualifier(*args, **kwargs)[source]
Bases:
ProtocolMarker for distinguishing multiple bindings of the same type.
Why this exists: One interface often has several implementations (primary vs replica DB). A qualifier +
Keyselects 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:
ClockReal 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
- class pyiv.RealConsole(stream: TextIO | None = None, stdin: TextIO | None = None)[source]
Bases:
BaseConsoleProduction 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
- 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_style() Dict[str, bool][source]
Get text style state (not available for RealConsole).
- Returns:
Dictionary with all styles False
- 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)
- 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
- 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
- class pyiv.RealFilesystem[source]
Bases:
FilesystemReal filesystem implementation using the standard library.
Use this in production when you need actual disk I/O. Prefer
MemoryFilesystemin 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)
- unlink(path: str | Path, missing_ok: bool = False) None[source]
Remove a file.
- Parameters:
path – File path to remove
missing_ok – Don’t raise error if file doesn’t exist
- class pyiv.ReflectionConfig[source]
Bases:
ConfigConfiguration 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, )
- 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:
TypeError – If interface is not a type
ImportError – If the package cannot be imported
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:
ProtocolProtocol 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)
- class pyiv.SerDe[source]
Bases:
ChainHandlerAbstract 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_typeinstead of hard-codingjson.dumps/pickle.loads. Register implementations on the ENCODING chain so injectors and pipelines can swap formats without rewriting call sites. Multiple handlers may share ahandler_typewith different behaviors (date formatting, null handling, and so on).Subclasses must implement
handler_type,serialize, anddeserialize.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
- class pyiv.ServiceCommand(args: Namespace, injector: Any | None = None)[source]
Bases:
CommandBase class for long-running service commands.
Use this when a command should run until interrupted (daemons, workers).
execute()runsinit()→run()→cleanup()and mapsKeyboardInterruptto 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
- 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
- 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'
- class pyiv.SingletonScope[source]
Bases:
objectPer-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())
- class pyiv.SingletonType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]
Bases:
EnumLegacy enum for singleton lifecycle on
Config.register.Why this exists: Older call sites use
singleton_type=...instead of aScope. PreferScopefor 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:
EnumInjector construction stage: lazy vs fail-fast singletons.
Why this exists: Lazy singletons hide binding mistakes until first use. Pass
stage=Stage.PRODUCTIONtoget_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:
ClockSynthetic 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
- class pyiv.Timer[source]
Bases:
ABCAbstract handle for a scheduled callback.
Use timers when code needs one-shot or repeating callbacks driven by a
Clock. PreferSyntheticClock.start_timerin tests so time advances only when you calladvance(), 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()
- class pyiv.XMLSerDe[source]
Bases:
SerDeXML 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
itemchildren. 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)
- 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