Python 3.3: yield from, venv, and Namespace Packages

# Log file processing pipeline

def read_lines(path):
    """Generator that reads a file line by line."""
    with open(path) as f:
        for line in f:
            yield line.strip()

def filter_errors(lines):
    """Keep only lines containing ERROR."""
    for line in lines:
        if "ERROR" in line:
            yield line

def extract_messages(lines):
    """Extract the message after the log level."""
    for line in lines:
        parts = line.split(" - ", maxsplit=2)
        if len(parts) >= 3:
            yield parts[2]

def error_pipeline(paths):
    """Full pipeline: read multiple files and extract errors.

    Thanks to yield from, each step naturally delegates
    to the sub-generator without an explicit loop.
    """
    for path in paths:
        lines = read_lines(path)
        errors = filter_errors(lines)
        yield from extract_messages(errors)

# Usage
log_files = ["app.log", "worker.log", "api.log"]
for message in error_pipeline(log_files):
    print(f"[ERROR] {message}")

# Another example: flatten a nested structure
def flatten(structure):
    """Recursively flatten a nested structure."""
    for element in structure:
        if isinstance(element, (list, tuple)):
            yield from flatten(element)
        else:
            yield element

data = [1, [2, 3], [4, [5, 6]], 7]
print(list(flatten(data)))  # [1, 2, 3, 4, 5, 6, 7]

# Creating a virtual environment from the terminal
# python -m venv my_project_env

# Activation (Linux/macOS)
# source my_project_env/bin/activate

# Activation (Windows)
# my_project_env\Scripts\activate

# Checking that the environment is active
import sys
import os

def check_environment():
    """Display information about the active Python environment."""
    venv_path = os.environ.get("VIRTUAL_ENV")
    if venv_path:
        print(f"Active virtual environment: {venv_path}")
        print(f"Python executable: {sys.executable}")
        print(f"Package paths:")
        for path in sys.path:
            if "site-packages" in path:
                print(f"  {path}")
    else:
        print("No active virtual environment.")

check_environment()

# Programmatic environment creation
import venv

builder = venv.EnvBuilder(with_pip=True)
builder.create("test_env")
print("Environment 'test_env' created with bundled pip.")

# Directory layout:
#
# directory_a/
#   myframework/
#     core/
#       __init__.py
#       engine.py
#
# directory_b/
#   myframework/
#     plugins/
#       __init__.py
#       csv_export.py
#
# No __init__.py in myframework/!
# Python 3.3 automatically merges both parts.

# In code, you import normally:
# import myframework.core.engine
# import myframework.plugins.csv_export

# Checking whether a package is a namespace package
import importlib

def inspect_package(name):
    """Display whether a package is a namespace package."""
    try:
        module = importlib.import_module(name)
        if hasattr(module, "__path__"):
            paths = list(module.__path__)
            if hasattr(module, "__file__") and module.__file__:
                print(f"{name}: regular package ({module.__file__})")
            else:
                print(f"{name}: namespace package")
                for path in paths:
                    print(f"  -> {path}")
        else:
            print(f"{name}: simple module")
    except ImportError:
        print(f"{name}: not found")

inspect_package("json")  # regular package
inspect_package("os")    # simple module

import faulthandler
import sys
import threading

# Enable faulthandler to capture segfaults
faulthandler.enable()

# Dump the call stack of all threads
# (useful for debugging a program that appears stuck)
def long_running_task():
    """Simulate a blocking task."""
    import time
    time.sleep(30)

thread = threading.Thread(target=long_running_task, name="BackgroundWork")
thread.daemon = True
thread.start()

# Display the state of all threads
faulthandler.dump_traceback(file=sys.stderr, all_threads=True)

# Trigger a dump after a delay (watchdog)
# If the program does not respond within 10 seconds,
# the call stack will be printed automatically.
faulthandler.dump_traceback_later(
    timeout=10,
    repeat=False,
    file=sys.stderr,
)

# You can also enable it via the command line:
# python -X faulthandler my_script.py
#
# Or via an environment variable:
# PYTHONFAULTHANDLER=1 python my_script.py

# Cancel the watchdog if everything went fine
faulthandler.cancel_dump_traceback_later()