Application startup utilities

Functions and context managers for application startup, environment loading, and worker process safety.

The alexber.utils.mains module provides utilities for common application startup tasks: fixing the working directory, loading .env files inside packed distributions, and protecting multiprocessing pools from unpicklable exceptions.

Working directory helpers

fixabscwd()

Changes the current working directory to the directory that contains the __main__ module.

When you launch a Python script from a different directory, relative file paths (config files, assets, etc.) resolve against the shell’s working directory instead of the script’s location. fixabscwd() corrects this once at startup.

import alexber.utils.mains as mains

# Call once, early in your entry-point
mains.fixabscwd()

# Now relative paths resolve from the directory that contains __main__
with open('config/settings.yml') as f:
    ...

fixabscwd() is a no-op when running inside a REPL, an IPython notebook, or a frozen executable (sys.frozen == True).

FixRelCwd(relPackage, logger=None)

A context manager that temporarily changes the working directory to the location of an installed package, then restores the original directory on exit.

Useful when you need to call a module that resolves relative paths against its own installation directory.

import myapp.worker as worker
from alexber.utils.mains import FixRelCwd

with FixRelCwd(worker) as directory:
    # CWD is now the directory where worker/__init__.py lives
    worker.run()  # relative paths inside worker resolve correctly

# CWD has been restored to its original value

Parameter	Type	Description
relPackage	module	The installed package whose directory will become the working directory.
logger	logging.Logger | None	Logger for debug output. Defaults to the module logger.

---

Worker process guard

GuardedWorkerException(logger=None, suppress=False, default_exc_message=...)

A context manager that prevents pool.join() from hanging indefinitely when a worker process raises an exception that cannot be pickled back to the parent.

Many exception types (for example, subprocess.CalledProcessError which holds stdout/stderr buffers) are not picklable. When such an exception escapes a multiprocessing.Pool worker, the pool’s join call blocks forever. GuardedWorkerException catches the exception, logs it, and optionally re-raises a plain, picklable Exception in its place.

This is a known CPython bug that still exists in Python 3. Always wrap the body of Pool worker functions with this context manager.

from multiprocessing import Pool
from alexber.utils.mains import GuardedWorkerException
import logging

log = logging.getLogger(__name__)

def worker_task(item):
    with GuardedWorkerException(logger=log):
        # Any unpicklable exception raised here is replaced with a plain Exception
        result = subprocess.run(['my-tool', item], check=True, capture_output=True)
        return result.stdout

with Pool(4) as pool:
    pool.map(worker_task, items)
    pool.join()  # will no longer hang

Parameter	Type	Default	Description
logger	logging.Logger | None	None	Logger to record the caught exception. If None, traceback is written to stderr.
suppress	bool	False	When True the caught exception is silently swallowed. When False a plain Exception is raised instead.
default_exc_message	str	"Worker failed"	Message used when re-raising the replacement exception.

---

Environment loading

load_env(**kwargs)

Loads a .env file into os.environ. Unlike calling python-dotenv directly, this function can locate .env files that are packaged inside eggs or other archive formats via the importlib.resources API.

1. Install the optional dependency

   pip install alex-ber-utils[python-dotenv]
   
2. Call load_env at application startup

   from alexber.utils.mains import load_env
   
   # Option 1 — path on disk
   load_env(dotenv_path='/path/to/.env')
   
   # Option 2 — stream (e.g. from a StringIO or vault secret)
   import io
   load_env(stream=io.StringIO('MY_VAR=hello'))
   
   # Option 3 — locate .env inside an installed package (works in eggs)
   load_env(ENV_PCK='myapp.config', ENV_NAME='.env')
   
   # Option 4 — no arguments; falls back to python-dotenv defaults
   load_env()
   

Parameter	Description
ENV_PCK	Dotted package name containing the .env file. Used with importlib.resources.
ENV_NAME	Filename of the .env file inside ENV_PCK. Defaults to ".env".
dotenv_path	Absolute or relative path to a .env file on disk.
stream	A StringIO or file-like object with .env content.

When both dotenv_path/stream and ENV_PCK are absent, all remaining kwargs are forwarded directly to python-dotenv’s load_dotenv().

fix_env(**kwargs)

Prepends the absolute path prefix of a reference package to selected os.environ entries.

Useful when the environment contains relative path fragments that must be resolved against the package’s installation directory at runtime.

from alexber.utils.mains import fix_env

# Suppose os.environ['PLUGIN_PATH'] = 'plugins/core'
# After fix_env, it becomes '/installed/path/to/myapp/plugins/core'
fix_env(
    ENV_KEYS='PLUGIN_PATH',
    ENV_MAIN_PCK='myapp',
)

Parameter	Description
ENV_KEYS	Comma-separated names of os.environ keys to fix.
ENV_MAIN_PCK	Package whose __init__.py directory defines the prefix.
ENV_KEY_SEP	Separator between key names. Default: ",".
ENV_SEP	Path separator used inside paths. Default: os.path.sep.
ENV_DELIM_SEP	Delimiter used inside an os.environ value when it contains multiple paths. Default: os.pathsep.
cls	Class or dotted string for the implementation. Default: OsEnvrionPathExpender.

fix_retry_env(**kwargs)

Fixes Windows-style paths (e.g. C:\\data\\file) stored in os.environ so they resolve correctly on Linux by stripping the drive letter.

from alexber.utils.mains import fix_retry_env

# os.environ['DATA_DIR'] = 'C:\\app\\data' (written on Windows)
fix_retry_env(ENV_KEYS='DATA_DIR')
# os.environ['DATA_DIR'] is now '/app/data'

Use fix_retry_env in CI pipelines or Docker images that consume environment variables originally set on a Windows developer machine.

---

Type-checking utilities

These helpers are used internally but are part of the public API.

is_iterable(value)

Returns True if value is iterable, explicitly excluding None, str, and bytes.

from alexber.utils.mains import is_iterable

is_iterable([1, 2, 3])   # True
is_iterable({'a': 1})    # True
is_iterable('hello')     # False
is_iterable(None)        # False
is_iterable(42)          # False

is_mapping(value)

Returns True if value is a mapping (has a callable .items attribute).

from alexber.utils.mains import is_mapping

is_mapping({'key': 'value'})  # True
is_mapping([('key', 'val')])  # False

make_hashable(obj)

Recursively converts an object to a hashable type so it can be used as a dictionary key or set element.

• Mappings become frozenset of (key, value) pairs.
• Iterables become tuple.
• Objects with __hash__ are returned unchanged.
• Everything else is wrapped in HashableWrapper.

from alexber.utils.mains import make_hashable

key = make_hashable({'env': 'prod', 'region': ['us', 'eu']})
cache = {key: result}  # works

HashableWrapper

A fallback wrapper that makes any object hashable by delegating __hash__ to hash(str(obj)).

from alexber.utils.mains import HashableWrapper

obj = SomeUnhashableClass()
wrapper = HashableWrapper(obj)
hash(wrapper)  # works
wrapper == HashableWrapper(obj)  # True if obj == obj

---

OsEnvrionPathExpender and OsEnvrionPathRetry

These classes implement the logic behind fix_env() and fix_retry_env() respectively. You can pass a custom subclass (or its dotted import path) via the cls parameter of those functions to override the default behaviour.

from alexber.utils.mains import fix_env
from myapp.patches import CustomPathExpender

fix_env(
    ENV_KEYS='PLUGIN_PATH',
    ENV_MAIN_PCK='myapp',
    cls=CustomPathExpender,   # or cls='myapp.patches.CustomPathExpender'
)