pyio-intercept 0.1.0

A generic stream interceptor for sys.stdout or sys.stderr. Allows teeing output to pluggable handlers.

pyio_intercept

A generic, transparent stream interceptor for Python's standard I/O (sys.stdout, sys.stderr, sys.stdin) and file operations (builtins.open).

pyio_intercept allows you to tee, block, modify, or audit terminal and file data using a powerful action middleware chain without disrupting the original streams or crashing your application.

Features

• Middleware Action Chains: Pass data through multiple action handlers. Actions can pass, modify, or completely block the payload.
• Comprehensive Intercepts: Supports sys.stdout, sys.stderr, sys.stdin, and builtins.open (files).
• Context Manager: Supports scoped interception (e.g., with StdoutIntercept():).
• Long-lived Mode: Persistent overriding via .install() and .uninstall().
• Thread Safe: Middleware chains are executed using isolated snapshot copies, ensuring safe concurrent writes from multiple threads.
• Exception Isolation: Middleware errors are caught and logged directly to sys.__stderr__ without crashing the main application flow.
• Context Aware: Middleware actions receive a context dictionary containing the active stream or file object being manipulated.

Installation

You can install this module directly using pip:

pip install .

For development (editable mode):

pip install -e .

The Middleware Pattern

All intercepts in pyio_intercept use an Action Chain. An action is simply a function that receives the current payload, a next_action callable to pass data down the chain, and a context dictionary containing metadata about the stream/file.

Calling Convention & Flow

1. payload: The data being read or written. For .write(), it's the string chunk. For .read(), it starts as None.
2. next_action(payload): The callback to the next middleware in the chain. You must return the result of next_action() if you want the data to continue flowing. If you return None (or omit it), the chain stops and the data is dropped.
3. context: A dictionary containing contextual metadata (like {'stream': '<stdout>'} or {'file': <file_object>}).
4. End of the Chain: The final next_action is automatically bound to the original, intercepted function (e.g., the real OS-level sys.__stdout__.write or the real file's .read()).

ASCII Diagram: How Data Flows

Write Operations (e.g., print() or file.write()): Data flows forward through the chain, transforming the payload before it hits the disk/terminal.

Application call: print("hello")
       │
       ▼ (payload = "hello")
[ Action 1: uppercase ] ───(modifies payload to "HELLO")──┐
                                                          │
       ┌──────────────────────────────────────────────────┘
       ▼ (payload = "HELLO")
[ Action 2: prefix ] ──────(modifies payload to "[LOG] HELLO")──┐
                                                                │
       ┌────────────────────────────────────────────────────────┘
       ▼ (payload = "[LOG] HELLO")
[ Original Function ] (e.g. sys.__stdout__.write)
       │
       ▼
   Terminal Output: "[LOG] HELLO"

Read Operations (e.g., file.read() or sys.stdin.read()): For reads, the call starts with a None payload and travels down the chain. The actual file/input content is retrieved at the bottom, and transformations are applied as the data flows back up the call stack to the application.

Request Flow (Call Stack)              Response Flow (Data Returning)
=========================              ==============================

  Application: file.read()               Application gets: "HELLO"
       │                                      ▲
       ▼ (starts with payload=None)           │ (returns "HELLO")
[ Action 1: uppercase ]                [ Action 1: uppercase ]
       │                                      ▲
       ▼ (calls next_action)                  │ (returns "hello")
[ Original Function ]                  [ Original Function ]
       │                                      ▲
       ▼ (reads from disk)                    │ (retrieves)
      Disk                                   Disk (contains "hello")

Examples

1. Basic Mirroring (Stdout)

Tee output to a file while still printing to the terminal.

from pyio_intercept import StdoutIntercept

def logger_action(payload, next_action, context):
    with open("app.log", "a") as f:
        f.write(f"LOG: {payload}\n")
    return next_action(payload)

with StdoutIntercept(actions=[logger_action]):
    print("This goes to the terminal AND app.log")

2. Content Blocking (Stdout)

Block specific sensitive data from ever reaching the console.

from pyio_intercept import StdoutIntercept

def block_secrets(payload, next_action, context):
    if "PASSWORD=" in payload:
        return None # Blocks the payload entirely
    return next_action(payload)

with StdoutIntercept(actions=[block_secrets]):
    print("Connecting to DB...")
    print("PASSWORD=super_secret_123") # Will not be printed!

3. Modifying Input Dynamically (Stdin)

Force all user input to be uppercase as it is read by the application.

from pyio_intercept import StdinIntercept
import sys

def force_uppercase(payload, next_action, context):
    # For reads, the payload travels BACK up the chain
    result = next_action(payload) 
    return result.upper() if result else result

with StdinIntercept(actions=[force_uppercase]):
    # Any input typed here by the user is returned as uppercase
    user_input = sys.stdin.readline() 

4. File I/O Audit Tracking (FileIntercept)

Track exactly when files are opened and how many lines are written to them.

import time
from collections import defaultdict
from pyio_intercept import FileIntercept

audit_log = defaultdict(lambda: {"open_time": None, "lines_written": 0})

def on_file_open(file_obj, *args, **kwargs):
    filename = getattr(file_obj, 'name', str(file_obj))
    audit_log[filename]["open_time"] = time.time()

def track_lines(payload, next_action, context):
    file_obj = context.get('file')
    if file_obj and isinstance(payload, str) and payload:
        lines = payload.count('\n')
        if lines > 0:
            audit_log[file_obj.name]["lines_written"] += lines
    return next_action(payload)

with FileIntercept(actions=[track_lines], on_open=on_file_open):
    with open("data.txt", "w") as f:
        f.write("Line 1\nLine 2\n")
        
# audit_log now contains the open time and the fact that 2 lines were written!

5. On-the-fly Read Modification (File IO)

Capitalize text automatically as it is read from the disk.

from pyio_intercept import FileIntercept

def capitalize_read(payload, next_action, context):
    result = next_action(payload)
    return result.capitalize() if isinstance(result, str) else result

with FileIntercept(actions=[capitalize_read]):
    with open("data.txt", "r") as f:
        content = f.read() # Data is loaded from disk, then capitalized

Author

Jon Allen © 2026

License

This project is licensed under the MIT License.