Utilities¶

TaipanStack provides observability utilities for production applications.

Cache¶

Intelligent Cache decorator.

Provides in-memory caching that respects the Result monad and TTL, ignoring caching for Err() results.

CacheDecorator ¶

Bases: Protocol

Protocol for the cache decorator.

cached ¶

cached(ttl: float, max_size: int = 1024) -> CacheDecorator

Cache the Ok() results of a function for a given TTL.

Err() results are not cached. Supports both async and sync functions. Implements LRU (Least Recently Used) eviction when max_size is reached.

PARAMETER	DESCRIPTION
`ttl`	Time to live in seconds. TYPE: `float`
`max_size`	Maximum number of elements to store in the cache. TYPE: `int` DEFAULT: `1024`

RETURNS	DESCRIPTION
`CacheDecorator`	Decorator function.

Context¶

Observability Context Module.

Provides context variables and context managers for tracing and observability, such as the correlation ID.

get_correlation_id ¶

get_correlation_id() -> str | None

Get the current correlation ID.

RETURNS	DESCRIPTION
`str \| None`	The correlation ID if set, otherwise None.

set_correlation_id ¶

set_correlation_id(correlation_id: str | None) -> None

Set the correlation ID for the current context.

PARAMETER	DESCRIPTION
`correlation_id`	The correlation ID string, or None to clear. TYPE: `str \| None`

correlation_scope ¶

correlation_scope(
    correlation_id: str | None,
) -> Iterator[None]

Context manager to set the correlation ID and restore it after.

PARAMETER	DESCRIPTION
`correlation_id`	The correlation ID to set for the duration of the scope. TYPE: `str \| None`

YIELDS	DESCRIPTION
`None`	None

Concurrency¶

Concurrency utilities.

Provides a bulkhead pattern concurrency limiter decorator for both synchronous and asynchronous functions. Uses an OverloadError and returns a Result type.

OverloadError ¶

OverloadError(message: str = 'Concurrency limit reached')

Bases: Exception

Exception raised when a concurrency limit is exceeded or timed out.

Initialize the OverloadError.

PARAMETER	DESCRIPTION
`message`	The error message to display. Defaults to "Concurrency limit reached". TYPE: `str` DEFAULT: `'Concurrency limit reached'`

ConcurrencyLimitDecorator ¶

Bases: Protocol

Protocol for the concurrency limit decorator.

limit_concurrency ¶

limit_concurrency(
    max_tasks: int, timeout: float = 0.0
) -> ConcurrencyLimitDecorator

Decorate a function to apply the bulkhead concurrency limit pattern.

If the maximum concurrent executions are reached, the wrapper will wait up to timeout seconds to acquire a execution slot. If it fails, it returns an Err(OverloadError).

PARAMETER	DESCRIPTION
`max_tasks`	Maximum concurrent function executions allowed. TYPE: `int`
`timeout`	Maximum time in seconds to wait for a slot if limit is reached. TYPE: `float` DEFAULT: `0.0`

RETURNS	DESCRIPTION
`ConcurrencyLimitDecorator`	Decorated function returning a `Result[T, OverloadError]`.

Example

@limit_concurrency(max_tasks=2, timeout=0.1) ... def process_data() -> str: ... return "data" process_data() Ok('data')

Filesystem¶

Safe filesystem operations.

Provides secure wrappers around file operations with path validation, atomic writes, and proper error handling using Result types.

FileNotFoundErr `dataclass` ¶

FileNotFoundErr(path: Path)

Error when file is not found.

message `property` ¶

message: str

Get the error message.

NotAFileErr `dataclass` ¶

NotAFileErr(path: Path)

Error when path is not a file.

message `property` ¶

message: str

Get the error message.

FileTooLargeErr `dataclass` ¶

FileTooLargeErr(path: Path, size: int, max_size: int)

Error when file exceeds size limit.

message `property` ¶

message: str

Get the error message.

WriteOptions `dataclass` ¶

WriteOptions(
    base_dir: Path | str | None = None,
    encoding: str = "utf-8",
    create_parents: bool = True,
    backup: bool = True,
    atomic: bool = True,
)

Options for safe_write.

ATTRIBUTE	DESCRIPTION
`base_dir`	Base directory to constrain to. TYPE: `Path \| str \| None`
`encoding`	File encoding. TYPE: `str`
`create_parents`	Create parent directories if needed. TYPE: `bool`
`backup`	Create backup of existing file. TYPE: `bool`
`atomic`	Use atomic write. TYPE: `bool`

safe_read ¶

safe_read(
    path: Path | str,
    *,
    base_dir: Path | str | None = None,
    encoding: str = "utf-8",
    max_size_bytes: int | None = 10 * 1024 * 1024,
) -> Result[str, ReadFileError]

Read a file safely with path validation.

PARAMETER	DESCRIPTION
`path`	Path to the file to read. TYPE: `Path \| str`
`base_dir`	Base directory to constrain to. TYPE: `Path \| str \| None` DEFAULT: `None`
`encoding`	File encoding. TYPE: `str` DEFAULT: `'utf-8'`
`max_size_bytes`	Maximum file size to read (None for no limit). TYPE: `int \| None` DEFAULT: `10 * 1024 * 1024`

RETURNS	DESCRIPTION
`Ok`	File contents on success. TYPE: `str`
`Err`	Error details on failure. TYPE: `ReadFileError`

Example

result = safe_read("config.json") if isinstance(result, Ok): ... data = json.loads(result.unwrap()) ... else: ... err = result.unwrap_err() ... if isinstance(err, FileNotFoundErr): ... print(f"Missing: {err.path}") ... elif isinstance(err, FileTooLargeErr): ... print(f"Too big: {err.size} bytes")

safe_write ¶

safe_write(
    path: Path | str,
    content: str,
    *,
    options: WriteOptions | None = None,
) -> Path

Write to a file safely with path validation.

PARAMETER	DESCRIPTION
`path`	Path to write to. TYPE: `Path \| str`
`content`	Content to write. TYPE: `str`
`options`	Write options. TYPE: `WriteOptions \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`Path`	Path to the written file.

RAISES	DESCRIPTION
`SecurityError`	If path validation fails.

ensure_dir ¶

ensure_dir(
    path: Path | str,
    *,
    base_dir: Path | str | None = None,
    mode: int = 493,
) -> Path

Ensure a directory exists, creating it if needed.

PARAMETER	DESCRIPTION
`path`	Path to the directory. TYPE: `Path \| str`
`base_dir`	Base directory to constrain to. TYPE: `Path \| str \| None` DEFAULT: `None`
`mode`	Directory permissions. TYPE: `int` DEFAULT: `493`

RETURNS	DESCRIPTION
`Path`	Path to the directory.

RAISES	DESCRIPTION
`SecurityError`	If path validation fails.
`FileExistsError`	If a file already exists at the given path or intermediate paths.

Logging¶

Structured logging with context.

Provides a configured logger with support for structured output, context propagation, and proper formatting.

StackLogger ¶

StackLogger(
    name: str = "stack",
    level: str = "INFO",
    *,
    use_structured: bool = False,
)

Enhanced logger with context support.

Provides a wrapper around standard logging with additional features like context propagation and structured output support.

ATTRIBUTE	DESCRIPTION
`name`	Logger name.
`level`	Current log level.

Initialize the logger.

PARAMETER	DESCRIPTION
`name`	Logger name. TYPE: `str` DEFAULT: `'stack'`
`level`	Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL). TYPE: `str` DEFAULT: `'INFO'`
`use_structured`	Use structured logging if structlog is available. TYPE: `bool` DEFAULT: `False`

bind ¶

bind(**context: object) -> StackLogger

Add context to logger.

PARAMETER	DESCRIPTION
`**context`	Key-value pairs to add to context. TYPE: `object` DEFAULT: `{}`

RETURNS	DESCRIPTION
`StackLogger`	Self for chaining.

unbind ¶

unbind(*keys: str) -> StackLogger

Remove context keys.

PARAMETER	DESCRIPTION
`*keys`	Keys to remove from context. TYPE: `str` DEFAULT: `()`

RETURNS	DESCRIPTION
`StackLogger`	Self for chaining.

debug ¶

debug(message: str, **kwargs: object) -> None

Log a debug message.

PARAMETER	DESCRIPTION
`message`	The message to log. TYPE: `str`
`**kwargs`	Additional context. TYPE: `object` DEFAULT: `{}`

info ¶

info(message: str, **kwargs: object) -> None

Log an info message.

PARAMETER	DESCRIPTION
`message`	The message to log. TYPE: `str`
`**kwargs`	Additional context. TYPE: `object` DEFAULT: `{}`

warning ¶

warning(message: str, **kwargs: object) -> None

Log a warning message.

PARAMETER	DESCRIPTION
`message`	The message to log. TYPE: `str`
`**kwargs`	Additional context. TYPE: `object` DEFAULT: `{}`

error ¶

error(message: str, **kwargs: object) -> None

Log an error message.

PARAMETER	DESCRIPTION
`message`	The message to log. TYPE: `str`
`**kwargs`	Additional context. TYPE: `object` DEFAULT: `{}`

critical ¶

critical(message: str, **kwargs: object) -> None

Log a critical message.

PARAMETER	DESCRIPTION
`message`	The message to log. TYPE: `str`
`**kwargs`	Additional context. TYPE: `object` DEFAULT: `{}`

exception ¶

exception(message: str, **kwargs: object) -> None

Log an exception with traceback.

PARAMETER	DESCRIPTION
`message`	The message to log. TYPE: `str`
`**kwargs`	Additional context. TYPE: `object` DEFAULT: `{}`

mask_sensitive_data_processor ¶

mask_sensitive_data_processor(
    _logger: object,
    _method: str,
    event_dict: MutableMapping[str, object],
) -> MutableMapping[str, object]

Mask sensitive data in structlog event dictionaries.

Intercept the event_dict produced by structlog and replace the value of any key whose name contains a sensitive substring with "***REDACTED***". Matching is case-insensitive.

PARAMETER	DESCRIPTION
`_logger`	The wrapped logger object (unused, required by structlog). TYPE: `object`
`_method`	The name of the log method called (unused, required by structlog). TYPE: `str`
`event_dict`	The structured event dictionary. TYPE: `MutableMapping[str, object]`

RETURNS	DESCRIPTION
`MutableMapping[str, object]`	The event dictionary with sensitive values masked.

correlation_id_processor ¶

correlation_id_processor(
    _logger: object,
    _method: str,
    event_dict: MutableMapping[str, object],
) -> MutableMapping[str, object]

Structlog processor to inject correlation ID into events.

PARAMETER	DESCRIPTION
`_logger`	The wrapped logger object. TYPE: `object`
`_method`	The name of the log method called. TYPE: `str`
`event_dict`	The structured event dictionary. TYPE: `MutableMapping[str, object]`

RETURNS	DESCRIPTION
`MutableMapping[str, object]`	The event dictionary with correlation_id injected if sets.

setup_logging ¶

setup_logging(
    level: Literal[
        "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"
    ] = "INFO",
    *,
    format_type: Literal[
        "simple", "detailed", "json"
    ] = "detailed",
    log_file: str | None = None,
    use_structured: bool = False,
) -> None

Configure the root logger.

PARAMETER	DESCRIPTION
`level`	Log level to set. TYPE: `Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']` DEFAULT: `'INFO'`
`format_type`	Output format type. TYPE: `Literal['simple', 'detailed', 'json']` DEFAULT: `'detailed'`
`log_file`	Optional file to log to. TYPE: `str \| None` DEFAULT: `None`
`use_structured`	Use structlog if available. TYPE: `bool` DEFAULT: `False`

get_logger ¶

get_logger(
    name: str = "stack",
    *,
    level: str = "INFO",
    use_structured: bool = False,
) -> StackLogger

Get a configured logger instance.

PARAMETER	DESCRIPTION
`name`	Logger name. TYPE: `str` DEFAULT: `'stack'`
`level`	Log level. TYPE: `str` DEFAULT: `'INFO'`
`use_structured`	Use structlog if available. TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`StackLogger`	Configured StackLogger instance.

Example

logger = get_logger("my_module") logger.bind(request_id="123").info("Processing request")

log_operation ¶

log_operation(
    operation: str,
    *,
    logger: StackLogger | None = None,
    level: str = "INFO",
    expected_exceptions: tuple[type[Exception], ...]
    | type[Exception] = Exception,
) -> AbstractContextManager[StackLogger]

Context manager for logging operations.

PARAMETER	DESCRIPTION
`operation`	Name of the operation. TYPE: `str`
`logger`	Logger to use (creates one if not provided). TYPE: `StackLogger \| None` DEFAULT: `None`
`level`	Log level for messages. TYPE: `str` DEFAULT: `'INFO'`
`expected_exceptions`	Exceptions to catch and log as failures. TYPE: `tuple[type[Exception], ...] \| type[Exception]` DEFAULT: `Exception`

YIELDS	DESCRIPTION
`AbstractContextManager[StackLogger]`	The logger instance.

Example

with log_operation("setup") as logger: ... logger.info("Setting up environment")

Rate Limit¶

Rate limiting utilities.

Provides an in-memory token-bucket based rate limiting decorator for both synchronous and asynchronous functions. The decorator returns a Result type encapsulating the original return value or a RateLimitError error.

RateLimitError ¶

RateLimitError(message: str = 'Rate limit exceeded')

Bases: Exception

Exception raised when a rate limit is exceeded.

Initialize the RateLimitError.

PARAMETER	DESCRIPTION
`message`	The error message to display.Defaults to "Rate limit exceeded". TYPE: `str` DEFAULT: `'Rate limit exceeded'`

RateLimiter ¶

RateLimiter(max_calls: int, time_window: float)

Token bucket rate limiter logic.

Initialize the token bucket.

PARAMETER	DESCRIPTION
`max_calls`	The maximum number of calls allowed in the time window. TYPE: `int`
`time_window`	The time window in seconds. TYPE: `float`

consume ¶

consume(tokens: float = 1.0) -> bool

Try to consume tokens.

PARAMETER	DESCRIPTION
`tokens`	Number of tokens to consume. Defaults to 1.0. TYPE: `float` DEFAULT: `1.0`

RETURNS	DESCRIPTION
`bool`	True if tokens were consumed (allow), False otherwise (limit exceeded).

RateLimitDecorator ¶

Bases: Protocol

Protocol for the rate limit decorator.

rate_limit ¶

rate_limit(
    max_calls: int, time_window: float
) -> RateLimitDecorator

Decorate a function to apply rate limiting.

If the rate limit is exceeded, the wrapped function immediately returns an Err(RateLimitError). Uses an in-memory token bucket strategy.

PARAMETER	DESCRIPTION
`max_calls`	Maximum function executions allowed in the defined window. TYPE: `int`
`time_window`	Time window size in seconds. TYPE: `float`

RETURNS	DESCRIPTION
`RateLimitDecorator`	Decorated function returning a `Result[T, RateLimitError]`.

Example

@rate_limit(max_calls=2, time_window=1.0) ... def fetch_data() -> str: ... return "data" fetch_data() Ok('data') fetch_data() Ok('data') fetch_data() Err(RateLimitError('Rate limit exceeded'))

Serialization¶

Serialization utilities.

Provides an optimized default encoder for use with orjson.dumps.

default_encoder ¶

default_encoder(obj: object) -> dict[str, object]

Default encoder for orjson.dumps handling Result types.

Intercepts objects of type Ok and Err: - Ok(value): Returns {"status": "success"} merged with value if value is a dict. Otherwise returns {"status": "success", "data": value}. - Err(error): Returns {"status": "error", "message": str(error)}.

PARAMETER	DESCRIPTION
`obj`	The object to encode. TYPE: `object`

RETURNS	DESCRIPTION
`dict[str, object]`	A serializable representation of the object.

RAISES	DESCRIPTION
`TypeError`	If the object type is not supported.

Example

import orjson orjson.dumps(Ok({"id": 1}), default=default_encoder) b'{"status":"success","id":1}' orjson.dumps(Err(ValueError("oops")), default=default_encoder) b'{"status":"error","message":"oops"}'

Subprocess¶

Safe subprocess execution with security guards.

Provides secure wrappers around subprocess execution with command validation, timeout handling, and retry logic.

SafeCommandResult `dataclass` ¶

SafeCommandResult(
    command: list[str],
    returncode: int,
    stdout: str = "",
    stderr: str = "",
    duration_seconds: float = 0.0,
)

Result of a safe command execution.

ATTRIBUTE	DESCRIPTION
`command`	The executed command. TYPE: `list[str]`
`returncode`	Exit code of the command. TYPE: `int`
`stdout`	Standard output. TYPE: `str`
`stderr`	Standard error. TYPE: `str`
`success`	Whether the command succeeded (returncode == 0). TYPE: `bool`
`duration_seconds`	How long the command took. TYPE: `float`

success `property` ¶

success: bool

Check if command succeeded.

raise_on_error ¶

raise_on_error() -> SafeCommandResult

Raise an exception if command failed.

RETURNS	DESCRIPTION
`SafeCommandResult`	Self if successful.

RAISES	DESCRIPTION
`CalledProcessError`	If command failed.

run_safe_command ¶

run_safe_command(
    command: Sequence[str],
    *,
    cwd: Path | str | None = None,
    timeout: float = 300.0,
    capture_output: bool = True,
    check: bool = False,
    allowed_commands: Sequence[str] | None = None,
    env: dict[str, str] | None = None,
    allowed_env_vars: Sequence[str] | None = None,
    dry_run: bool = False,
) -> SafeCommandResult

Execute a command safely with security guards.

This function provides a secure wrapper around subprocess.run with command injection protection, timeout handling, and optional command whitelisting.

PARAMETER	DESCRIPTION
`command`	Command and arguments as a sequence. TYPE: `Sequence[str]`
`cwd`	Working directory for the command. TYPE: `Path \| str \| None` DEFAULT: `None`
`timeout`	Maximum execution time in seconds. TYPE: `float` DEFAULT: `300.0`
`capture_output`	Whether to capture stdout/stderr. TYPE: `bool` DEFAULT: `True`
`check`	Whether to raise on non-zero exit. TYPE: `bool` DEFAULT: `False`
`allowed_commands`	Whitelist of allowed commands. TYPE: `Sequence[str] \| None` DEFAULT: `None`
`env`	Environment variables to set. TYPE: `dict[str, str] \| None` DEFAULT: `None`
`allowed_env_vars`	Whitelist of allowed environment variables. TYPE: `Sequence[str] \| None` DEFAULT: `None`
`dry_run`	If True, don't actually execute the command. TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`SafeCommandResult`	SafeCommandResult with execution details.

RAISES	DESCRIPTION
`SecurityError`	If command validation fails.
`TimeoutExpired`	If command times out.
`CalledProcessError`	If check=True and command fails.

Example

result = run_safe_command(["poetry", "install"]) if result.success: ... print("Installation complete!")

Utilities¶

Cache¶

CacheDecorator ¶

cached ¶

Context¶

get_correlation_id ¶

set_correlation_id ¶

correlation_scope ¶

Concurrency¶

OverloadError ¶

ConcurrencyLimitDecorator ¶

limit_concurrency ¶

Filesystem¶

FileNotFoundErr dataclass ¶

message property ¶

NotAFileErr dataclass ¶

message property ¶

FileTooLargeErr dataclass ¶

message property ¶

WriteOptions dataclass ¶

safe_read ¶

safe_write ¶

ensure_dir ¶

Logging¶

StackLogger ¶

bind ¶

unbind ¶

debug ¶

info ¶

warning ¶

error ¶

critical ¶

exception ¶

mask_sensitive_data_processor ¶

correlation_id_processor ¶

setup_logging ¶

get_logger ¶

log_operation ¶

Rate Limit¶

RateLimitError ¶

RateLimiter ¶

consume ¶

RateLimitDecorator ¶

rate_limit ¶

Serialization¶

default_encoder ¶

Subprocess¶

SafeCommandResult dataclass ¶

success property ¶

raise_on_error ¶

run_safe_command ¶

FileNotFoundErr `dataclass` ¶

message `property` ¶

NotAFileErr `dataclass` ¶

message `property` ¶

FileTooLargeErr `dataclass` ¶

message `property` ¶

WriteOptions `dataclass` ¶

SafeCommandResult `dataclass` ¶

success `property` ¶