Resolvers Module

The resolvers package provides lazy value computation for TreeStore nodes.

Module Structure

        graph TB
    subgraph "genro_treestore.resolvers"
        BASE[base.py<br/>TreeStoreResolver]
        CB[callback.py<br/>CallbackResolver]
        DIR[directory.py<br/>DirectoryResolver]
        TXT[txtdoc.py<br/>TxtDocResolver]

        BASE --> CB
        BASE --> DIR
        BASE --> TXT
    end

TreeStoreResolver

class genro_treestore.TreeStoreResolver[source]

Bases: object

Base class for lazy/dynamic value resolution.

A resolver computes a node’s value on-demand instead of storing it statically. Subclasses must implement the load() method.

There are two main use cases:

Traversal resolvers: Return a TreeStore that enables further path navigation. The resolved TreeStore is stored in node._value to allow traversal to continue. Use cache_time=-1 for infinite caching with manual invalidation via reset().
Leaf resolvers: Return a scalar value (temperature, computed result, etc.). With read_only=True and cache_time=0, each access recomputes the value without storing it.

parent_node: The TreeStoreNode this resolver is attached to. Set automatically when assigning resolver to a node.

cache_time: Cache duration in seconds. 0=no cache, >0=seconds, -1=infinite (until reset() is called).

read_only: If True, resolved value is not stored in node._value. Only applies when accessing node.value directly, not during path traversal (which always populates _value).

Example

Subclass and implement load():

class MyResolver(TreeStoreResolver):
    @smartasync
    async def load(self):
        return await fetch_data()

node.resolver = MyResolver(cache_time=300)
value = node.value  # Triggers load(), caches for 5 minutes

Parameters:

cache_time (int (default: 0))
read_only (bool (default: True))
kwargs (Any)

__init__(cache_time=0, read_only=True, **kwargs)[source]

Initialize the resolver.

Parameters:

cache_time (int (default: 0)) –
Cache duration in seconds. - 0: No caching, always recompute (default). - >0: Cache result for this many seconds. - -1: Cache forever until reset() is called. Recommended

for traversal resolvers with manual invalidation.
read_only (bool (default: True)) – If True, resolved value is not stored in node._value when accessing node.value directly. Note: during path traversal, _value is always populated regardless of this setting, to enable navigation through the resolved TreeStore.
**kwargs (Any) – Additional arguments stored in _init_kwargs for serialization. Subclasses should store their custom parameters here.

Return type:

None

parent_node: TreeStoreNode | None

read_only

property cache_time: int: Cache duration in seconds (0=none, >0=seconds, -1=infinite).

property expired: bool

Check if the cached value has expired.

Returns:: True if cache is expired, not present, or caching is disabled. False if cache is still valid.

reset()[source]

Invalidate the cache, forcing recomputation on next access.

For traversal resolvers with cache_time=-1, call this method when the underlying data source has changed to trigger a fresh load() on the next access.

Example

>>> resolver = DirectoryResolver('/path/to/dir', cache_time=-1)
>>> store.set_resolver('docs', resolver)
>>> store['docs.file.txt']  # Loads directory contents
>>> # ... files change on disk ...
>>> resolver.reset()  # Invalidate cache
>>> store['docs.file.txt']  # Reloads directory contents

Return type:: None

load()[source]

Load and return the resolved value.

Subclasses must override this method to provide the actual value computation. The method is decorated with @smartasync, enabling transparent sync/async operation:

From sync context: executed via asyncio.run() automatically
From async context: returns coroutine for normal await

For traversal resolvers, return a TreeStore containing the hierarchical data. For leaf resolvers, return a scalar value.

Return type:: Any
Returns:: The resolved value. For traversal resolvers, this should be a TreeStore instance. For leaf resolvers, any value type.
Raises:: NotImplementedError – If not overridden in subclass.

Example

Async resolver fetching remote data:

@smartasync
async def load(self):
    async with aiohttp.ClientSession() as session:
        async with session.get(self.url) as response:
            data = await response.json()
    return TreeStore(data)

Sync resolver reading local files:

@smartasync
async def load(self):
    # Can be sync - smartasync handles it
    with open(self.path) as f:
        return f.read()

serialize()[source]

Serialize the resolver for persistence/export.

Creates a dictionary containing all information needed to recreate the resolver via deserialize(). Based on Genropy’s BagResolver.resolverSerialize() pattern.

The serialized data includes: - resolver_module: Full module path (e.g., ‘myapp.resolvers’) - resolver_class: Class name (e.g., ‘DirectoryResolver’) - args: Positional arguments from _init_args - kwargs: Keyword arguments including cache_time, read_only,

and any custom kwargs from _init_kwargs

For CallbackResolver, the callback function must be a top-level importable function (not a lambda or nested function) for serialization to work correctly.

Return type:: dict[str, Any]
Returns:: Dictionary suitable for JSON serialization and later reconstruction via deserialize().

Example

>>> resolver = DirectoryResolver('/path', cache_time=-1)
>>> data = resolver.serialize()
>>> # data = {
>>> #     'resolver_module': 'genro_treestore.resolver',
>>> #     'resolver_class': 'DirectoryResolver',
>>> #     'args': ('/path',),
>>> #     'kwargs': {'cache_time': -1, 'read_only': True}
>>> # }

classmethod deserialize(data)[source]

Recreate a resolver from serialized data.

Dynamically imports the resolver class and instantiates it with the stored arguments.

Parameters:

data (dict[str, Any]) – Dictionary from serialize() containing resolver_module, resolver_class, args, and kwargs.

Return type:

TreeStoreResolver

Returns:

New resolver instance of the appropriate subclass.

Raises:

ModuleNotFoundError – If resolver_module cannot be imported.
AttributeError – If resolver_class is not found in the module.

Example

>>> data = {
...     'resolver_module': 'genro_treestore.resolver',
...     'resolver_class': 'CallbackResolver',
...     'args': (my_callback,),
...     'kwargs': {'cache_time': 300}
... }
>>> resolver = TreeStoreResolver.deserialize(data)

CallbackResolver

class genro_treestore.CallbackResolver[source]

Bases: TreeStoreResolver

Resolver that invokes a callback function to compute values.

A simple resolver that delegates value computation to a user-provided function. The callback receives the parent node as its argument, allowing access to the node’s context (label, attributes, siblings).

This is useful for: - Computed values based on other nodes in the tree - Simple transformations or calculations - Quick prototyping before creating a custom resolver class

For serialization to work, the callback must be a top-level function that can be imported by name (not a lambda or nested function).

callback: The function to call on each resolution.

Example

Computed value based on siblings:

def compute_total(node):
    store = node.parent
    price = store.get_item('price')
    quantity = store.get_item('quantity')
    return price * quantity

store.set_item('price', 100)
store.set_item('quantity', 5)
store.set_item('total')
store.set_resolver('total', CallbackResolver(compute_total))

print(store['total'])  # 500

With caching:

resolver = CallbackResolver(
    expensive_computation,
    cache_time=300  # Cache for 5 minutes
)

Parameters:

callback (Callable[[TreeStoreNode], Any])
kwargs (Any)

__init__(callback, **kwargs)[source]

Initialize the callback resolver.

Parameters:

callback (Callable[[TreeStoreNode], Any]) – Function to invoke for value computation. Signature: callback(node: TreeStoreNode) -> Any The node parameter is the TreeStoreNode this resolver is attached to, providing access to node.parent (the containing TreeStore), node.label, node.attr, etc.
**kwargs (Any) – Additional arguments passed to TreeStoreResolver. Common options: cache_time, read_only.

Return type:

None

Example

>>> def get_timestamp(node):
...     return datetime.now().isoformat()
>>> resolver = CallbackResolver(get_timestamp, cache_time=0)

callback

load()[source]

Invoke the callback and return its result.

Return type:: Any
Returns:: The value returned by self.callback(self.parent_node).

Example

from genro_treestore import TreeStore, CallbackResolver

store = TreeStore()
store.set_item('config.base', 'https://api.example.com')
store.set_item('config.version', 'v2')

def compute_url(node):
    parent = node.parent
    return f"{parent['base']}/{parent['version']}"

store.set_item('config.full_url')
store.set_resolver('config.full_url', CallbackResolver(compute_url))

print(store['config.full_url'])  # 'https://api.example.com/v2'

DirectoryResolver

class genro_treestore.DirectoryResolver[source]

Bases: TreeStoreResolver

Resolver for lazy loading of filesystem directory contents.

Loads directory contents on demand, creating a TreeStore where: - Subdirectories become branch nodes with their own DirectoryResolver - Files become leaf nodes (value=None by default, or processed content) - File metadata stored in node attributes

Compatible with Genropy’s DirectoryResolver API.

path: Absolute path to the directory.

relocate: Relative path prefix for rel_path attribute.

invisible: If True, include hidden files (starting with ‘.’).

ext: Comma-separated list of extensions to process (e.g., ‘xml,json’). Format: ‘ext’ or ‘ext:processor_name’.

include: Glob pattern for files to include (e.g., ‘*.py’).

exclude: Glob pattern for files/dirs to exclude (e.g., ‘__pycache__’).

callback: Optional callback(nodeattr) -> bool to filter nodes.

dropext: If True, don’t include extension in node labels.

processors: Dict mapping extension to processor function.

Example

Basic directory listing:

resolver = DirectoryResolver('/home/user/docs')
store.set_item('docs')
store.set_resolver('docs', resolver)

# Access triggers lazy load
for label in store['docs'].keys():
    node = store.get_node(f'docs.{label}')
    print(f"{label}: {node.attr['file_ext']}")

With XML processing:

resolver = DirectoryResolver(
    '/path/to/config',
    ext='xml',
    processors={'xml': lambda p: parse_xml(p)}
)

Parameters:

path (str)
relocate (str (default: ''))
cache_time (int (default: 500))
read_only (bool (default: True))
invisible (bool (default: False))
ext (str (default: ''))
include (str (default: ''))
exclude (str (default: ''))
callback (Optional[Callable[[dict], bool | None]] (default: None))
dropext (bool (default: False))
processors (dict[str, Union[Callable[[str], Any], bool]] | None (default: None))
kwargs (Any)

__init__(path, relocate='', *, cache_time=500, read_only=True, invisible=False, ext='', include='', exclude='', callback=None, dropext=False, processors=None, **kwargs)[source]

Initialize the directory resolver.

Parameters:

path (str) – Absolute path to the directory to resolve.
relocate (str (default: '')) – Relative path prefix for rel_path attribute. Used to maintain relative paths when nested.
cache_time (int (default: 500)) – Cache duration in seconds. Default 500 (like Bag).
read_only (bool (default: True)) – If True, resolved value not stored in node._value.
invisible (bool (default: False)) – If True, include hidden files (starting with ‘.’).
ext (str (default: '')) – Comma-separated extensions to process. Format: ‘xml’ or ‘xml:processor_name,json:processor_name’.
include (str (default: '')) – Glob pattern for files to include (e.g., ‘.py,.txt’).
exclude (str (default: '')) – Glob pattern for files/dirs to exclude.
callback (Optional[Callable[[dict], bool | None]] (default: None)) – Function called with nodeattr dict for each entry. Return False to skip the entry, None/True to include.
dropext (bool (default: False)) – If True, don’t include extension in node labels.
processors (dict[str, Union[Callable[[str], Any], bool]] | None (default: None)) – Dict mapping extension/processor_name to: - Callable[[str], Any]: Function to process file, returns value - False: Skip files with this extension - None: Use default processor (returns None)
**kwargs (Any) – Additional arguments for TreeStoreResolver.

Return type:

None

path

relocate

invisible

ext

include

exclude

callback

dropext

processors

property instance_kwargs: dict[str, Any]: Get kwargs for creating child DirectoryResolvers.

load()[source]

Load directory contents into a TreeStore.

Return type:: TreeStore
Returns:: TreeStore containing directory entries as nodes. Subdirectories have DirectoryResolver attached. Files have metadata in attributes.

processor_directory(path)[source]

Process a subdirectory by creating a new DirectoryResolver.

Parameters:: path (str) – Absolute path to the subdirectory.
Return type:: DirectoryResolver
Returns:: DirectoryResolver for the subdirectory.

processor_default(path)[source]

Default processor for files - returns None.

Parameters:: path (str) – Absolute path to the file.
Return type:: None
Returns:: None (file path stored in attributes).

Example

from genro_treestore import TreeStore, DirectoryResolver

store = TreeStore()
store.set_item('files')
store.set_resolver('files', DirectoryResolver('/path/to/dir'))

# Directory contents loaded on first access
for name in store['files'].keys():
    print(name)

TxtDocResolver

class genro_treestore.TxtDocResolver[source]

Bases: TreeStoreResolver

Resolver that loads text file contents.

Compatible with Genropy’s TxtDocResolver.

path: Absolute path to the text file.

Example

>>> resolver = TxtDocResolver('/path/to/file.txt')
>>> node.resolver = resolver
>>> content = node.value  # Reads file contents

Parameters:

path (str)
cache_time (int (default: 500))
read_only (bool (default: True))
kwargs (Any)

__init__(path, *, cache_time=500, read_only=True, **kwargs)[source]

Initialize the text document resolver.

Parameters:

path (str) – Absolute path to the text file.
cache_time (int (default: 500)) – Cache duration in seconds. Default 500.
read_only (bool (default: True)) – If True, resolved value not stored in node._value.
**kwargs (Any) – Additional arguments for TreeStoreResolver.

Return type:

None

path

load()[source]

Load and return the file contents as bytes.

Return type:: bytes
Returns:: File contents as bytes.

Example

from genro_treestore import TreeStore, TxtDocResolver

store = TreeStore()
store.set_item('readme')
store.set_resolver('readme', TxtDocResolver('README.md'))

print(store['readme'])  # File contents

Resolver Architecture

        classDiagram
    class TreeStoreResolver {
        <<abstract>>
        +load(node)*
        +cache_time: int
    }

    class CallbackResolver {
        +callback: Callable
        +cache_time: int
        +load(node)
    }

    class DirectoryResolver {
        +path: Path
        +load(node)
    }

    class TxtDocResolver {
        +path: Path
        +load(node)
    }

    TreeStoreResolver <|-- CallbackResolver
    TreeStoreResolver <|-- DirectoryResolver
    TreeStoreResolver <|-- TxtDocResolver

Resolver Lifecycle

        stateDiagram-v2
    [*] --> Attached: set_resolver()
    Attached --> Loading: value accessed
    Loading --> Cached: load() returns
    Cached --> Loading: cache expired
    Cached --> [*]: node deleted

Caching

Resolver	Default Cache	Configurable
`CallbackResolver`	None	Yes (`cache_time`)
`DirectoryResolver`	Permanent	No
`TxtDocResolver`	Permanent	No

Creating Custom Resolvers

from genro_treestore import TreeStoreResolver

class ApiResolver(TreeStoreResolver):
    """Fetch value from REST API."""

    def __init__(self, endpoint: str, cache_time: int = 60):
        self.endpoint = endpoint
        self.cache_time = cache_time

    def load(self, node):
        import requests
        response = requests.get(self.endpoint)
        return response.json()

# Usage
store.set_resolver('api.data', ApiResolver('https://api.example.com/data'))

Resolvers Module

Module Structure

TreeStoreResolver

CallbackResolver

Example

DirectoryResolver

Example

TxtDocResolver

Example

Resolver Architecture

Resolver Lifecycle

Caching

Creating Custom Resolvers

See Also