Skip to main content

Directed inputs class consumes and processes inputs from sources beyond args and kwargs

Project description

Directed Inputs Class

Directed Inputs Class Logo

🛠️ Manage your Python inputs efficiently! 🎯

CI Status Documentation Status PyPI Package latest release Supported versions

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_inputs on 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_config for 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:

  1. Explicit argument - service.method(domain="example.com") always wins
  2. Stdin JSON - If from_stdin=True and key matches parameter name
  3. Environment variable - If from_env=True (checks DOMAIN or env_prefix + DOMAIN)
  4. 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

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

directed_inputs_class-202511.7.0.tar.gz (19.4 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

directed_inputs_class-202511.7.0-py3-none-any.whl (14.0 kB view details)

Uploaded Python 3

File details

Details for the file directed_inputs_class-202511.7.0.tar.gz.

File metadata

File hashes

Hashes for directed_inputs_class-202511.7.0.tar.gz
Algorithm Hash digest
SHA256 b1ba4aef1fb35a88ad5a87050cbbb2ea6f118b38a520f012e5b4680d03e32f7e
MD5 a74015623994fe068a6c0419c1640487
BLAKE2b-256 6c47f0c58dbaf33458e090a4c5d328a502efd9af8ee57645370dcfec19f8303b

See more details on using hashes here.

File details

Details for the file directed_inputs_class-202511.7.0-py3-none-any.whl.

File metadata

File hashes

Hashes for directed_inputs_class-202511.7.0-py3-none-any.whl
Algorithm Hash digest
SHA256 1f34ac4807806cc3fc0c20d75b4ac858b9fd7f7759de69e05664702fa26f0133
MD5 8a68e23189da651e87b6b4f58dbd8719
BLAKE2b-256 3ccd16874885ae65b0549c8e7426aaf89ff3c2467a7e650da53573e753b319f4

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page