Module

utils.thread_local

Thread-local caching utilities.

Provides a generic thread-local cache for expensive-to-create objects like parsers, database connections, or pipeline instances.

Example:

from bengal.utils.thread_local import ThreadLocalCache

# Create a cache for markdown parsers
parser_cache = ThreadLocalCache(
    factory=lambda: create_markdown_parser(),
    name="markdown_parser",
)

# Get or create parser for current thread
parser = parser_cache.get()

# Get parser with a specific key (e.g., engine type)
mistune_parser = parser_cache.get("mistune")

Related Modules:

  • bengal.rendering.pipeline.thread_local: Uses for parser caching
  • bengal.orchestration.render: Uses for pipeline caching
  • bengal.core.site.core: Uses for rendering context caching
View source
2 Classes

Classes

ThreadLocalCache
Generic thread-local cache with factory pattern. Creates one instance per thread per key, reusing …
5

Generic thread-local cache with factory pattern.

Creates one instance per thread per key, reusing it for subsequent calls. Useful for expensive objects like parsers that are not thread-safe but can be reused within a single thread.

Thread Safety:

Each thread gets its own instance(s), no locking required for access.
The factory function should be thread-safe if it accesses shared state.

Performance:

  • First access per thread/key: factory() cost (e.g., 10ms for parser)
  • Subsequent access: ~1µs (attribute lookup)
Inherits from Generic[T]

Methods 3

get
Get or create cached instance for current thread.
1 T 
def get(self, key: str | None = None) -> T

Get or create cached instance for current thread.

Parameters 1
key str | None

Optional key for multiple instances per thread. Use when caching different variants (e.g., parser engines).
Returns

T —

Cached or newly created instance

clear
Clear cached instance for current thread.
1 None 
def clear(self, key: str | None = None) -> None

Clear cached instance for current thread.

Parameters 1
key str | None

Specific key to clear, or None to clear default
clear_all
Clear all cached instances for current thread.
0 None 
def clear_all(self) -> None

Clear all cached instances for current thread.

Internal Methods 2
__init__
Initialize thread-local cache.
2 None 
def __init__(self, factory: Callable[[], T] | Callable[[str], T], name: str = 'default')

Initialize thread-local cache.

Parameters 2
factory Callable[[], T] | Callable[[str], T]

Callable that creates new instances. Can be no-arg or accept a key string.
name str

Name for this cache (used in attribute names)
_check_factory_signature
Check if factory accepts a key argument.
0 bool 
def _check_factory_signature(self) -> bool

Check if factory accepts a key argument.

Returns

bool

ThreadSafeSet
Thread-safe set for tracking created resources (e.g., directories).
5

Thread-safe set for tracking created resources (e.g., directories).

Methods 2

add_if_new
Add item if not present, return True if added. Thread-safe check-and-add operation.
1 bool 
def add_if_new(self, item: str) -> bool

Add item if not present, return True if added.

Thread-safe check-and-add operation.

Parameters 1
item str

Item to add
Returns

bool —

True if item was new (added), False if already present

clear
0 None 
def clear(self) -> None
Internal Methods 3
__init__
0 None 
def __init__(self) -> None
__contains__
1 bool 
def __contains__(self, item: str) -> bool
Parameters 1
item str
Returns

bool

__len__
0 int 
def __len__(self) -> int
Returns

int