aeth-ext

This markdown file and subclass_searchengine.py was AI generated by Claude. All other code was written by me. Sweet Fire Tobacco shared library — a batteries-included foundation for building Python services, batch jobs, and CLI tools.

aeth-ext bundles the cross-cutting infrastructure that the Sweet Fire Tobacco projects rely on: a one-call application bootstrap, an opinionated logging stack built on Rich, pydantic-based settings, FTP/SFTP transfer adapters, a static (import-free) subclass discovery engine, a monkey-patching framework, alert emails, and assorted utilities.

• Python: >=3.14
• Package name (PyPI): aeth-ext
• Import name: aeth_ext
• Author: Jacob Ogden

Table of contents

• aeth-ext
  • Table of contents
  • Installation
  • Quick start
  • Architecture overview
  • Modules
    • aeth_ext — application bootstrap
    • settings — configuration
      • Helpers
    • logging — logging stack
      • Constant auto-configuration
      • Subclassing
      • Public API
    • errors — fatal-exception handling & alerts
    • ftp — FTP / SFTP adapters
      • Shared interface
      • Errors
    • monkey_patcher — patch framework
    • subclass_searchengine — static class discovery
    • const_parsing — constant extraction
    • utils — email & datetime helpers
    • types — shared types & mixins
    • rich — enhanced progress bars

Installation

The library is published to the internal SFTPyPI index.

# base install
uv add aeth-ext

# with the high-performance async event loop (uvloop on Linux, winloop on Windows)
uv add "aeth-ext[async]"

# with SFTP support (paramiko)
uv add "aeth-ext[sftp]"

# everything
uv add "aeth-ext[async,sftp]"

Core runtime dependencies: aiologic, pydantic-settings, python-dateutil, rich, tzdata.

Quick start

initialize() belongs inside your if __name__ == "__main__": block — not inside a helper function. The logging auto-configuration reads uppercase constants (e.g. PROJECT_NAME, RICH_CONSOLE) directly out of __main__ via AST — they are captured and evaluated in isolation, so their position relative to initialize() within the block does not matter.

# myapp/__main__.py
if __name__ == "__main__":
    from sys import platform
    from rich.console import Console
    from aeth_ext import initialize

    RICH_CONSOLE = Console(
        width=None if platform == "win32" else 165,
        log_time=platform == "win32",
    )
    PROJECT_NAME = "MyApp"
    LOGGING_TYPE = "daily"

    initialize(asyncio=True)
else:
    # When this module is imported rather than run directly, get_console()
    # returns the same console object that was configured from RICH_CONSOLE
    # by initialize() — so this is not a fallback, it's the real thing.
    from rich import get_console
    RICH_CONSOLE = get_console()


# Rest of your application code below the guard
import asyncio
from logging import getLogger
...

The else branch is optional but recommended when other modules in your package import RICH_CONSOLE from __main__. Because initialize() updates Rich's global console in-place with the RICH_CONSOLE constant, get_console() returns that exact same object — so there's no difference between the two branches at runtime.

Architecture overview

A central theme of the library is static, import-free discovery: rather than forcing you to register components in a central place, aeth_ext scans your source tree, finds the most-derived subclass of a given base, and wires it up automatically. This powers settings (BaseSettings), logging (BaseLoggingConfig), and patches (MonkeyPatcher).

flowchart TD
    A["initialize()"] --> B["init_logging() / init_logging_worker()"]
    A --> C["MonkeyPatcher.apply_monkey_patches()"]
    A --> D["install uvloop / winloop (asyncio=True)"]
    B --> E["discover deepest BaseLoggingConfig subclass"]
    C --> F["discover MonkeyPatcher subclasses"]
    E --> G["subclass_searchengine"]
    F --> G
    H["BaseSettings.get_settings()"] --> I["CapturesSubclasses mixin"]

Modules

aeth_ext — application bootstrap

The package root exposes a single orchestration function.

def initialize(
    *queues: QueueCatchall,
    asyncio: bool = False,
    worker: bool = False,
    run_monkey_patches: bool = True,
    return_wrapped: bool = False,
) -> None | Callable[[], None]: ...

# Run immediately
initialize()

# Defer execution (e.g. to pass into a process pool initializer)
init = initialize(asyncio=True, return_wrapped=True)
init()

settings — configuration

BaseSettings extends pydantic_settings.BaseSettings and the CapturesSubclasses mixin, so the most-derived subclass is resolved automatically. In debug builds it reads a .env file; in release builds it relies purely on environment variables.

from aeth_ext.settings import BaseSettings


class Settings(BaseSettings):
    api_token: str  # required, from API_TOKEN


settings = Settings.get_settings()  # singleton; same instance every call

Key built-in fields (all overridable via env vars / .env):

Helpers

• get_settings() / get_final_model() — resolve the singleton.

logging — logging stack

A Rich-powered logging system with daily/per-run file rotation, abbreviated library paths, and queue-based fan-in for multi-process apps. initialize() calls init_logging() for you; you rarely need to call it directly.

Constant auto-configuration

init_logging discovers the deepest BaseLoggingConfig subclass, then inspects the parameter names of its configure_logging_main method. Each parameter name is uppercased to produce a constant name (e.g. project_name → PROJECT_NAME), and init_logging searches for matching uppercase assignments first in your running __main__ module and, if any values are still missing, in your project's __main__.py entrypoint script. Constants are evaluated with sys, platform, and Console available in the eval namespace.

The default configure_logging_main reads the following constants:

A required parameter with no matching constant and no default raises ValueError at startup. If you add parameters to a configure_logging_main override, those parameters are picked up from __main__ the same way.

Subclassing

from aeth_ext.logging.config import BaseLoggingConfig
from rich.console import Console


class LoggingConfig(BaseLoggingConfig):
    @classmethod
    def configure_logging_main(cls, rich_console: Console, project_name: str, **kw) -> None:
        super().configure_logging_main(rich_console=rich_console, project_name=project_name, **kw)
        # add extra handlers here
        ...

Public API

• init_logging(*queues) — main-process setup.
• init_logging_worker(queue) — worker-process setup; routes all log records to the parent process via a QueueHandler.
• BaseLoggingConfig — override configure_logging_main, configure_logging_worker, configure_base_per_runner, or configure_base_once to customise the stack.
• get_global_log_receiver() — returns the shared QueueHandler used for log fan-in; raises RuntimeError if not yet initialised.
• get_preferred_logrecord_formatter() / set_preferred_logrecord_formatter() — get or replace the shared formatter used by all file handlers.

errors — fatal-exception handling & alerts

Decorators that wrap a callable, log + email on any unhandled exception, set a shared FATAL_EVENT, and swallow the error (returning None).

from aeth_ext.errors.err_handling import (
    handle_fatal_exc_sync,
    handle_fatal_exc_async,
    FATAL_EVENT,
)


@handle_fatal_exc_sync
def risky() -> int:
    return 1 / 0  # logs, emails an alert, sets FATAL_EVENT, returns None


@handle_fatal_exc_async
async def risky_async() -> None:
    ...

send_alert_email(subject, content) composes and batch-sends an alert email to settings.alerts_recipients, attaching content as a UTF-8 file. It logs (and no-ops) if no recipients are configured.

ftp — FTP / SFTP adapters

AdaptedFTP and AdaptedSFTP expose an identical interface regardless of the underlying protocol. Code written against either adapter can switch between FTP and SFTP without modification — the protocol differences (binary-mode negotiation, SSL unwrapping, Paramiko vs ftplib internals) are fully encapsulated. Both adapters are context managers that open and close the connection automatically.

from aeth_ext.ftp.adapter import AdaptedFTP, AdaptedSFTP
from aeth_ext.rich.progress import Progress

# Exactly the same call-site whether ftp_or_sftp is AdaptedFTP or AdaptedSFTP
def process(ftp_or_sftp: AdaptedFTP | AdaptedSFTP) -> None:
    with ftp_or_sftp as conn:
        conn.download_file("/remote/report.csv", write_to_disk, task_msg="Downloading")
        conn.upload_file("/remote/out.csv", read_from_disk, file_size, task_msg="Uploading")

# Optional progress bar — works the same for both
with Progress() as pbar:
    with AdaptedSFTP(sftp_protocol, "my-server", pbar=pbar) as conn:  # or AdaptedFTP
        ok = conn.transfer_file("/src/file.csv", "/dst/file.csv", other_conn, task_msg="Relaying")

Shared interface

Both adapters implement every method listed below:

All methods assert that the adapter is open (i.e. used inside with) and raise AssertionError otherwise.

Errors

• ServerNotAvailableError(ConnectionError) — raised when a server is unreachable.

monkey_patcher — patch framework

Organize monkey patches as subclasses. Each plain method you define is forced into a staticmethod by the metaclass and is invoked once when patches are applied. The class is not instantiable — call its classmethods directly.

from aeth_ext.monkey_patcher import MonkeyPatcher


class MyPatches(MonkeyPatcher):
    def patch_some_library():
        import some_library
        some_library.thing = replacement


MonkeyPatcher.apply_monkey_patches()  # discovers + runs every subclass's patches

initialize(run_monkey_patches=True) calls apply_monkey_patches() for you.

subclass_searchengine — static class discovery

The engine behind the auto-wiring. It scans .py files with the ast module — without importing them — to find subclasses, then loads only the ones you ask for.

• find_subclasses(base, roots, *, ignored_dirs=..., include_name_fallback=False, recursive=True) → tuple[SubclassInfo, ...]
• get_entrypoint_root() → topmost package dir of the running entrypoint.
• SubclassInfo — NamedTuple with qualname, name, module, file, lineno, depth; call .load() to import the live class.
• iter_python_files, build_subclass_index, load_subclasses, reset_subclass_caches — supporting helpers.

const_parsing — constant extraction

Read uppercase constant assignments out of a source file via AST and safely evaluate them against a restricted namespace.

from pathlib import Path
from aeth_ext.const_parsing import parse_and_grab_constants

values = parse_and_grab_constants(
    Path("config.py"),
    expected_constants={"PROJECT_NAME": "project_name"},
    eval_locals={},
)
# -> {"project_name": "<value of PROJECT_NAME>"}

It scans both module-level statements and the if __name__ == "__main__": block.

utils — email & datetime helpers

Email composition / batch sending plus offset-aware datetime helpers.

from aeth_ext.utils import prepare_email_message, batch_send_emails, get_now, today

msg = prepare_email_message({
    "subject": "Report",
    "body": "See attached.",
    "from_addr": "[email protected]",
    "to_addrs": ["[email protected]"],
    "attachments": Path("report.csv"),
})
batch_send_emails(msg)  # SMTP config defaults to the alerts.* settings

types — shared types & mixins

• StrEnum — string enum whose value mirrors the member name.
• CapturesSubclasses (types/abc.py) — mixin that registers instances and can resolve the deepest subclass / final model; the backbone of the auto-wiring used by BaseSettings and BaseLoggingConfig.
• SingletonType, SingletonTypeABC, SingletonTypeBaseModel — singleton metaclasses.

rich — enhanced progress bars

Progress is a rich.progress.Progress subclass preconfigured with a sensible column layout (bar, M-of-N, percentage, time remaining). Its TaskID supports use as a context manager so a task is auto-removed on exit.

from aeth_ext.rich.progress import Progress

with Progress() as progress:
    with progress.add_task("Working", total=100) as task_id:
        progress.update(task_id, advance=50)
    # task is removed automatically here