Directed inputs class consumes and processes inputs from sources beyond args and kwargs
Project description
Directed Inputs Class
🛠️ Manage your Python inputs efficiently! 🎯
Directed Inputs Class provides flexible, transparent input handling for Python applications. Load inputs from environment variables, stdin, or dictionaries and have them automatically injected into your method arguments.
Key Features
- 🎯 Decorator-Based API - No inheritance required, use
@directed_inputson any class - 🧩 Automatic Type Coercion - String inputs converted to
bool,int,float,datetime,Path - 📥 Multi-Source Loading - Environment variables, stdin JSON, or predefined dictionaries
- 🧭 Scoped Environment Loading - Filter by prefix with optional stripping
- 🔄 Advanced Decoding - Base64, JSON, and YAML decoding with error handling
- ❄️ Input Freezing/Thawing - Immutable snapshots with restore capability
- 🔧 Per-Method Configuration - Use
@input_configfor custom parameter handling
Quick Start
Installation
pip install directed-inputs-class
Decorator API (Recommended)
The decorator API provides transparent input handling without inheritance:
from directed_inputs_class import directed_inputs, input_config
@directed_inputs(from_stdin=True, from_env=True)
class MyService:
"""Methods automatically receive inputs from stdin/environment."""
def list_users(self, domain: str | None = None, limit: int = 100) -> list:
# 'domain' and 'limit' are automatically populated from:
# 1. Stdin JSON (if from_stdin=True)
# 2. Environment variables (if from_env=True)
# 3. Default values (if not found)
return self._fetch_users(domain, limit)
@input_config("api_key", source_name="API_KEY", required=True)
def secure_operation(self, api_key: str, data: dict) -> dict:
# 'api_key' MUST be provided (from env var API_KEY)
# 'data' is loaded normally
return self._process(api_key, data)
# Usage
service = MyService()
result = service.list_users() # domain/limit loaded automatically
Input Resolution Order
For each method parameter, inputs are resolved in this order:
- Explicit argument -
service.method(domain="example.com")always wins - Stdin JSON - If
from_stdin=Trueand key matches parameter name - Environment variable - If
from_env=True(checksDOMAINorenv_prefix + DOMAIN) - Default value - Falls back to parameter default
Decorator Reference
@directed_inputs
Class decorator that enables automatic input handling for all methods:
@directed_inputs(
from_stdin=False, # Read JSON from stdin on first method call
from_env=True, # Load matching environment variables
env_prefix=None, # Filter env vars by prefix (e.g., "MY_APP_")
strip_env_prefix=False # Strip prefix from keys (MY_APP_FOO → FOO)
)
class MyService:
...
@input_config
Method decorator for per-parameter configuration:
@input_config(
"param_name", # Parameter to configure
source_name=None, # Alternative source name (e.g., "API_KEY" → api_key param)
aliases=None, # List of alternative names to check
required=False, # Raise error if not found
default=None, # Override default value
decode_base64=False, # Decode value from base64
decode_json=False, # Parse value as JSON
decode_yaml=False, # Parse value as YAML
)
def method(self, param_name: str):
...
InputContext
Access the input context directly:
@directed_inputs
class MyService:
def debug(self):
# Access raw inputs
ctx = self._input_context
print(f"All inputs: {ctx.inputs}")
print(f"Frozen: {ctx.frozen}")
# Manual input retrieval
value = ctx.get("key", default="fallback")
Type Coercion
String inputs from environment/stdin are automatically coerced based on type hints:
| Type Hint | Coercion |
|---|---|
bool |
"true"/"1"/"yes" → True, "false"/"0"/"no" → False |
int |
"42" → 42 |
float |
"3.14" → 3.14 |
Path |
"/tmp/file" → Path("/tmp/file") |
datetime |
ISO format string → datetime object |
dict |
JSON string → parsed dict |
list |
JSON string → parsed list |
str | None |
Handles Optional types correctly |
Example:
import os
from directed_inputs_class import directed_inputs
os.environ["DEBUG"] = "true"
os.environ["PORT"] = "8080"
@directed_inputs
class Config:
def get_settings(self, debug: bool = False, port: int = 3000) -> dict:
return {"debug": debug, "port": port}
config = Config()
print(config.get_settings()) # {"debug": True, "port": 8080}
Advanced Examples
Required Parameters
from directed_inputs_class import directed_inputs, input_config
@directed_inputs
class SecureService:
@input_config("api_key", required=True)
def call_api(self, api_key: str, endpoint: str) -> dict:
# Raises ValueError if api_key not in env/stdin
...
Base64 Decoding
import os
import base64
from directed_inputs_class import directed_inputs, input_config
# Set encoded value
os.environ["CERT"] = base64.b64encode(b"certificate-data").decode()
@directed_inputs
class TLSClient:
@input_config("cert", decode_base64=True)
def connect(self, cert: str) -> None:
# cert is automatically decoded from base64
print(cert) # "certificate-data"
Environment Prefix
import os
from directed_inputs_class import directed_inputs
os.environ["MY_APP_DATABASE_URL"] = "postgres://..."
os.environ["MY_APP_DEBUG"] = "true"
os.environ["OTHER_VAR"] = "ignored"
@directed_inputs(env_prefix="MY_APP_", strip_env_prefix=True)
class AppConfig:
def get_db(self, database_url: str | None = None) -> str:
# Only MY_APP_* variables loaded
# Prefix stripped: MY_APP_DATABASE_URL → DATABASE_URL
return database_url
Stdin JSON Input
echo '{"user": "alice", "count": 5}' | python script.py
from directed_inputs_class import directed_inputs
@directed_inputs(from_stdin=True)
class Processor:
def process(self, user: str, count: int = 10) -> str:
return f"Processing {count} items for {user}"
p = Processor()
print(p.process()) # "Processing 5 items for alice"
Legacy API (DirectedInputsClass)
The original inheritance-based API is still supported for backward compatibility:
from directed_inputs_class import DirectedInputsClass
class MyService(DirectedInputsClass):
def __init__(self):
super().__init__(from_environment=True, from_stdin=True)
def get_user(self, user_id: str | None = None) -> dict:
user_id = self.get_input("user_id", user_id)
return self._fetch_user(user_id)
Legacy Methods
| Method | Purpose |
|---|---|
get_input(name, default) |
Get input with fallback |
decode_input(name, ...) |
Get and decode (base64/json/yaml) |
freeze_inputs() |
Get immutable snapshot |
thaw_inputs() |
Restore from frozen state |
merge_inputs(dict) |
Deep merge additional inputs |
Migration to Decorator API
# Before (inheritance + manual get_input)
class OldService(DirectedInputsClass):
def method(self, domain: str | None = None):
domain = self.get_input("domain", domain)
return self._process(domain)
# After (decorator + automatic injection)
@directed_inputs
class NewService:
def method(self, domain: str | None = None):
return self._process(domain)
Integration with python-terraform-bridge
The decorator API integrates seamlessly with python-terraform-bridge for Terraform external data sources:
from directed_inputs_class import directed_inputs
from python_terraform_bridge import TerraformRegistry
registry = TerraformRegistry()
@directed_inputs(from_stdin=True) # Terraform passes inputs via stdin
class GitHubConnector:
@registry.data_source(key="repos", module_class="github")
def list_repos(self, org: str, include_private: bool = True) -> dict:
# Both decorators compose naturally:
# - @directed_inputs handles input loading from Terraform's stdin
# - @registry.data_source handles Terraform module generation
return {"repos": [...]}
API Reference
Exports
from directed_inputs_class import (
# Decorator API (recommended)
directed_inputs, # Class decorator
input_config, # Method decorator
InputConfig, # Configuration dataclass
InputContext, # Runtime context
# Legacy API (backward compatible)
DirectedInputsClass,
)
InputConfig Fields
| Field | Type | Default | Description |
|---|---|---|---|
name |
str |
required | Parameter name |
source_name |
str | None |
None |
Alternative env var name |
aliases |
list[str] |
[] |
Alternative names to check |
required |
bool |
False |
Raise if not found |
default |
Any |
None |
Override default |
decode_base64 |
bool |
False |
Decode from base64 |
decode_json |
bool |
False |
Parse as JSON |
decode_yaml |
bool |
False |
Parse as YAML |
Contributing
Contributions are welcome! Please see the Contributing Guidelines for more information.
Project Links
- Get Help (use the directed-inputs-class tag)
- PyPI
- GitHub
- Documentation
- Changelog
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file directed_inputs_class-202511.7.0.tar.gz.
File metadata
- Download URL: directed_inputs_class-202511.7.0.tar.gz
- Upload date:
- Size: 19.4 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.1.0 CPython/3.13.7
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
b1ba4aef1fb35a88ad5a87050cbbb2ea6f118b38a520f012e5b4680d03e32f7e
|
|
| MD5 |
a74015623994fe068a6c0419c1640487
|
|
| BLAKE2b-256 |
6c47f0c58dbaf33458e090a4c5d328a502efd9af8ee57645370dcfec19f8303b
|
File details
Details for the file directed_inputs_class-202511.7.0-py3-none-any.whl.
File metadata
- Download URL: directed_inputs_class-202511.7.0-py3-none-any.whl
- Upload date:
- Size: 14.0 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.1.0 CPython/3.13.7
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
1f34ac4807806cc3fc0c20d75b4ac858b9fd7f7759de69e05664702fa26f0133
|
|
| MD5 |
8a68e23189da651e87b6b4f58dbd8719
|
|
| BLAKE2b-256 |
3ccd16874885ae65b0549c8e7426aaf89ff3c2467a7e650da53573e753b319f4
|