datazen.environment package

Submodules

datazen.environment.base module

datazen - A base class to be extended for runtime data loading and storing.

class datazen.environment.base.BaseEnvironment(default_ns: str = '__root__', logger: ~logging.Logger = <Logger datazen.environment.base (WARNING)>, newline: str = '\n', **_)

Bases: object

The base class for environment loading-and-storing management.

add_dir(dir_type: DataType, dir_path: str, rel_path: str = '.', name: str = '__root__', allow_dup: bool = False) → bool

Proxy for add_dir for a namespace.

add_dirs(dir_type: DataType, dir_paths: List[str], rel_path: str = '.', name: str = '__root__', allow_dup: bool = False) → int

Proxy for add_dirs for a namespace.

add_namespace(name: str, clone_root: bool = True) → None

Add a new namespace, optionally clone from the existing root.

get_data(dir_type: DataType, name: str = '__root__') → Dict[str, Any]

Get the raw data for a directory type from a namespace.

get_namespace(key_name: str, target: str, target_data: Dict[str, Any]) → str

Determine the namespace that a target should use, in general they all should be unique unless they don’t load anything new.

get_to_load(dir_type: DataType, name: str = '__root__') → List[Path | str | None]

Proxy for a namespace’s get_to_load.

get_valid(name: str = '__root__') → bool

Get the ‘valid’ flag for a namespace.

set_valid(value: bool, name: str = '__root__') → None

Set the ‘valid’ flag for a namespace.

unload_all(name: str = '__root__') → None

Unload all of the directories for a namespace.

update_load_state(dir_type: DataType, to_load: List[Path | str | None], name: str = '__root__') → int

Proxy for update_load_state for a namespace.

class datazen.environment.base.Task(variant: str, name: str)

Bases: NamedTuple

Parameters identifying a task.

name: str

Alias for field number 1

property slug: str

Convert this task into its ‘slug’ form.

variant: str

Alias for field number 0

class datazen.environment.base.TaskResult(success: bool, fresh: bool, time_ns: int = -1)

Bases: NamedTuple

Return value for a task, express whether or not the task succeeded and if this tasks’ result should be considered ‘new’ from the last time it was evaluated.

fresh: bool

Alias for field number 1

log(task: Task, logger: Logger, level: int = 10) → None

Log status based on this task result.

success: bool

Alias for field number 0

time_ns: int

Alias for field number 2

with_time(time_ns: int) → TaskResult

Create a new result from this one, with the time set.

datazen.environment.base.dep_slug_unwrap(slug: str, default_op: str) → Task

From a slug String, determine the operation + target pair, if the operation isn’t specified use a default.

datazen.environment.command module

datazen - An environment extension that exposes command-line command execution.

class datazen.environment.command.CommandEnvironment(**kwargs)

Bases: TaskEnvironment

Exposes command-line commanding capability to the environment.

valid_command(entry: ~typing.Dict[str, ~typing.Any], _: str, __: ~typing.Dict[str, ~typing.Any] = None, deps_changed: ~typing.List[str] = None, logger: ~logging.Logger = <Logger datazen.environment.command (WARNING)>) → TaskResult

Perform the command specified by the entry.

datazen.environment.compile module

datazen - An environment extension that exposes compilation capabilities.

class datazen.environment.compile.CompileEnvironment(**kwargs)

Bases: TaskEnvironment

Leverages a cache-equipped environment to perform compilations.

valid_compile(entry: ~typing.Dict[str, ~typing.Any], namespace: str, dep_data: ~typing.Dict[str, ~typing.Any] = None, deps_changed: ~typing.List[str] = None, logger: ~logging.Logger = <Logger datazen.environment.compile (WARNING)>) → TaskResult

Perform the compilation specified by the entry.

datazen.environment.config module

datazen - A child class for adding configuration-data loading capabilities to

the environment dataset.

class datazen.environment.config.ConfigEnvironment(**kwargs)

Bases: VariableEnvironment, SchemaEnvironment

The configuration-data loading environment mixin, requires variable loading to function.

add_config_dirs(dir_paths: List[str], rel_path: str = '.', name: str = '__root__', allow_dup: bool = False) → int

Add configuration-data directories, return the number of directories added.

load_configs(cfg_loads: ~datazen.load.LoadedFiles = (None, None), var_loads: ~datazen.load.LoadedFiles = (None, None), sch_loads: ~datazen.load.LoadedFiles = (None, None), sch_types_loads: ~datazen.load.LoadedFiles = (None, None), name: str = '__root__', logger: ~logging.Logger = <Logger datazen.environment.config (WARNING)>, enforce_schemas: bool = True) → LoadResult

Load configuration data, resolve any un-loaded configuration directories.

datazen.environment.group module

datazen - A target-type for grouping other target-tasks together.

class datazen.environment.group.GroupEnvironment(**kwargs)

Bases: TaskEnvironment

Leverages a task-environment to group tasks together.

valid_group(entry: ~typing.Dict[str, ~typing.Any], _: str, dep_data: ~typing.Dict[str, ~typing.Any] = None, deps_changed: ~typing.List[str] = None, logger: ~logging.Logger = <Logger datazen.environment.group (WARNING)>) → TaskResult

Stub task to group other tasks.

datazen.environment.integrated module

datazen - A centralized store for runtime data.

class datazen.environment.integrated.Environment(newline: str = '\n')

Bases: CompileEnvironment, RenderEnvironment, GroupEnvironment, CommandEnvironment

A wrapper for inheriting all environment-loading capabilities.

command(target: str) → TaskResult

Execute a named ‘command’ target from the manifest.

compile(target: str) → TaskResult

Execute a named ‘compile’ target from the manifest.

execute(target: str = '', should_cache: bool = True) → TaskResult

Execute an arbitrary target.

execute_targets(targets: List[str]) → bool

Execute a list of named targets and return whether or not the entire set was successful.

group(target: str) → TaskResult

Attempt to satisfy a ‘group’ target.

render(target: str) → TaskResult

Execute a named ‘render’ target from the manifest.

datazen.environment.integrated.from_manifest(manifest_path: str, newline: str = '\n', data_cache_name: str = 'task_data', logger: ~logging.Logger = <Logger datazen.environment.integrated (WARNING)>) → Environment

Load an environment object from a schema definition on disk.

datazen.environment.manifest module

datazen - A class for adding manifest-loading to environments.

class datazen.environment.manifest.ManifestEnvironment(**kwargs)

Bases: ConfigEnvironment, TemplateEnvironment

A wrapper for the manifest-loading implementations of an environment.

default_target() → str

Retrieve a configured default target for a manifest, otherwise an empty String.

load_dirs(data: Dict[str, Any], rel_path: str, namespace: str = '__root__', allow_dup: bool = False, load_defaults: bool = True) → None

Looks for keys matching types of directories that can be loaded into an environment and tries to load them.

load_manifest(path: str = 'manifest.yaml', logger: ~logging.Logger = <Logger datazen.environment.manifest (WARNING)>) → bool

Attempt to load manifest data from a file.

load_manifest_reent(path: str, manifest_dir: str, params: ~typing.Dict[~typing.Any, ~typing.Any], files: ~typing.List[str], logger: ~logging.Logger = <Logger datazen.environment.manifest (WARNING)>) → Tuple[Dict[str, Any], bool]

Load a manifest recursively by resolving includes and merging the results.

path_fields = ['configs', 'schemas', 'schema_types', 'variables', 'templates']

targets_with_paths = ['compiles', 'renders']

static update_task_dirs(data: Dict[str, Any], rel_path: str) → None

Update path definitions in the provided manifest chunk to respect a relative path (e.g. the directory that the manifest chunk was loaded from).

datazen.environment.manifest.get_manifest_schema(require_all: bool = True, cls: ~typing.Type[~vcorelib.schemas.base.Schema] = <class 'vcorelib.schemas.CerberusSchema'>) → Schema

Load the schema for manifest from the package.

datazen.environment.manifest.get_output_dir(data: Dict[str, Any], rel_path: str, default: str = 'datazen-out') → str

Get the resolved output directory based on a dictionary containing target data.

datazen.environment.manifest.set_output_dir(data: Dict[str, Any], rel_path: str, default: str = 'datazen-out') → None

Set the ‘output_dir’ key correctly on a dictionary.

datazen.environment.manifest.update_path_relativity(path_strs: List[str], rel_path: str) → None

For a list of paths, if paths aren’t absolute, apply the provided relative path to the beginning of the path.

datazen.environment.manifest.validate_manifest(manifest: ~typing.Dict[str, ~typing.Any], logger: ~logging.Logger = <Logger datazen.environment.manifest (WARNING)>) → bool

Validate manifest data against the package schema.

datazen.environment.manifest_cache module

datazen - A class for adding caching to the manifest-loading environment.

class datazen.environment.manifest_cache.ManifestCacheEnvironment(**kwargs)

Bases: ManifestEnvironment

A wrapper for the cache functionality for an environment.

cached_enforce_schemas(data: Dict[str, Any], require_all: bool = True, name: str = '__root__') → bool

Enforce schemas, proxied through the cache.

cached_load_configs(name: str = '__root__', enforce_schemas: bool = True) → LoadResult

Load configs, proxied through the cache.

cached_load_schemas(require_all: bool = True, name: str = '__root__') → Dict[str, Any]

Load schemas, proxied through the cache.

cached_load_templates(name: str = '__root__') → Dict[str, Template]

Load templates, proxied through the cache.

cached_load_variables(name: str = '__root__') → LoadResult

Load variables, proxied through the cache.

clean_cache(purge_data: bool = True) → None

Remove cached data from the file-system.

describe_cache() → None

Describe the [initial] cache for debugging purposes.

get_new_loaded(types: List[str], load_checks: Dict[str, List[str]] = None) → int

Compute the number of new files loaded (since the initial load) for a set of types;

load_manifest_with_cache(path: str = 'manifest.yaml', logger: ~logging.Logger = <Logger datazen.environment.manifest_cache (WARNING)>) → bool

Load a manifest and its cache, or set up a new cache if one doesn’t exist.

restore_cache() → None

Return the cache to its initially-loaded state.

write_cache() → None

Commit cached data to the file-system.

datazen.environment.manifest_cache.manifest_cache_dir(path: str, manifest: Dict[str, Any]) → str

Find a manifest cache (path) from its path and data.

datazen.environment.render module

datazen - An environment extension that exposes rendering capabilities.

class datazen.environment.render.RenderEnvironment(**kwargs)

Bases: TaskEnvironment

Leverages a cache-equipped environment to render templates.

perform_render(template: ~jinja2.environment.Template, all_templates: ~typing.Dict[str, ~jinja2.environment.Template], path: str | None, entry: ~typing.Dict[str, ~typing.Any], data: ~typing.Dict[str, ~typing.Any] = None, logger: ~logging.Logger = <Logger datazen.environment.render (WARNING)>) → TaskResult

Render a template to the requested path using the provided data.

store_render(entry: Dict[str, Any], data: Dict[str, Any]) → None

Store data from the current render.

valid_render(entry: ~typing.Dict[str, ~typing.Any], namespace: str, dep_data: ~typing.Dict[str, ~typing.Any] = None, deps_changed: ~typing.List[str] = None, logger: ~logging.Logger = <Logger datazen.environment.render (WARNING)>) → TaskResult

Perform the render specified by the entry.

datazen.environment.render.get_render_children(children: Dict[str, Any], dep_data: Dict[str, Any], default_op: str, indent: int, delimeter: str = '', newline: str = '\n') → None

Build child dependency data.

datazen.environment.render.get_render_str(template: Template, name: str, indent: int, data: Dict[str, Any] = None, out_data: Dict[str, Any] = None, newline: str = '\n') → str

Render a template.

datazen.environment.render.indent_str(data: str, indent: int, newline: str = '\n') → str

Attempt to indent String data by some amount, based on some separator.

datazen.environment.render.render_name_to_key(name: str) → str

Convert the name of a render target with a valid dictionary key.

datazen.environment.schema module

datazen - A child class for adding schema-data loading capabilities to the

environment dataset.

class datazen.environment.schema.SchemaEnvironment(default_ns: str = '__root__', logger: ~logging.Logger = <Logger datazen.environment.base (WARNING)>, newline: str = '\n', **_)

Bases: BaseEnvironment

The schema-data loading environment mixin, only requires the base environment capability to function.

add_schema_dirs(dir_paths: List[str], rel_path: str = '.', name: str = '__root__', allow_dup: bool = False) → int

Add schema-data directories, return the number of directories added.

add_schema_type_dirs(dir_paths: List[str], rel_path: str = '.', name: str = '__root__', allow_dup: bool = False) → int

Add directories containing schema-type data (to be registered at runtime).

enforce_schemas(data: Dict[str, Any], require_all: bool = True, sch_loads: LoadedFiles = (None, None), sch_types_loads: LoadedFiles = (None, None), name: str = '__root__') → bool

Perform schema-validation on provided data and return the boolean result. Adds (and removes) namespaced types if applicable.

load_schema_types(sch_loads: LoadedFiles = (None, None), name: str = '__root__') → Dict[str, Any]

Load custom schema types and resolve any un-loaded directories.

load_schemas(require_all: bool = True, sch_loads: LoadedFiles = (None, None), sch_types_loads: LoadedFiles = (None, None), name: str = '__root__', modify_registry: bool = True) → Dict[str, Any]

Load schema data, resolve any un-loaded schema directories.

datazen.environment.task module

datazen - A class for exposing the capability to run concrete tasks against

the current environment.

class datazen.environment.task.TaskEnvironment(**kwargs)

Bases: ManifestCacheEnvironment

Adds capability for running tasks (including tasks with dependencies) and tracking whether or not updates are required.

already_satisfied(target: str, output_path: str | None, load_deps: ~typing.List[str], deps_changed: ~typing.List[str] = None, load_checks: ~typing.Dict[str, ~typing.List[str]] = None, logger: ~logging.Logger = <Logger datazen.environment.task (WARNING)>) → bool

Check if a target is already satisfied, if not debug-log some information about why not.

clean_cache(purge_data: bool = True) → None

Remove cached data from the file-system.

get_dep_data(dep_list: ~typing.List[str], logger: ~logging.Logger = <Logger datazen.environment.task (WARNING)>) → Dict[str, Any]

From a list of dependencies, create a dictionary with any task data they’ve saved.

get_manifest_entry(category: str, name: str) → Dict[str, Any]

Get an entry from the manifest.

handle_task(key_name: str, target: str, task_stack: List[Task] = None, should_cache: bool = True) → TaskResult

Handle the setup for manifest tasks, such as loading additional data directories and setting the correct output directory.

init_cache(cache_dir: str) → None

Initialize the task-data cache.

is_resolved(operation: str, target: str) → bool

Determine whether a target for an operation has already been resolved.

is_task_new(operation: str, target: str) → bool

Determine if a target has been newly executed this iteration.

push_dep(dep: str, task_stack: List[Task], curr_target: str) → None

Push a dependency onto a stack if they need to be resolved.

resolve(operation: str, target: str, should_cache: bool, is_new: bool) → None

Set a target for an operation as resolved.

resolve_dependencies(dep_list: ~typing.List[str], task_stack: ~typing.List[~datazen.environment.base.Task], target: str, logger: ~logging.Logger = <Logger datazen.environment.task (WARNING)>) → Tuple[bool, Dict[str, Any], List[str]]

Execute the entire chain of dependencies for a task, return the aggregate result as a boolean as well as the dependency data and which dependencies changed.

property task_data: Dict[str, Any]

Proxy task data through the cache.

valid_noop(entry: ~typing.Dict[str, ~typing.Any], _: str, __: ~typing.Dict[str, ~typing.Any] = None, ___: ~typing.List[str] = None, logger: ~logging.Logger = <Logger datazen.environment.task (WARNING)>) → TaskResult

Default handle for a potentially-unregistered operation.

write_cache() → None

Commit cached data to the file-system.

datazen.environment.task.get_dep_list(entry: Dict[str, Any]) → List[str]

From task data, build a list of task dependencies.

datazen.environment.task.get_path(entry: Dict[str, Any], key: str = 'name') → str

Get the full path to a render output from the manifest entry.

datazen.environment.template module

datazen - A child class for adding template-loading capabilities to the

environment dataset.

class datazen.environment.template.TemplateEnvironment(default_ns: str = '__root__', logger: ~logging.Logger = <Logger datazen.environment.base (WARNING)>, newline: str = '\n', **_)

Bases: BaseEnvironment

The template-loading environment mixin, only requires the base environment capability to function.

add_template_dirs(dir_paths: List[str], rel_path: str = '.', name: str = '__root__', allow_dup: bool = False) → int

Add template directories, return the number of directories added.

load_templates(template_loads: LoadedFiles = (None, None), name: str = '__root__') → Dict[str, Template]

Load templates, resolve any un-loaded template directories.

datazen.environment.variable module

datazen - A child class for adding variable-data loading capabilities to the

environment dataset.

class datazen.environment.variable.VariableEnvironment(default_ns: str = '__root__', logger: ~logging.Logger = <Logger datazen.environment.base (WARNING)>, newline: str = '\n', **_)

Bases: BaseEnvironment

The variable-data loading environment mixin. Only requires the base environment capability to function.

add_variable_dirs(dir_paths: List[str], rel_path: str = '.', name: str = '__root__', allow_dup: bool = False) → int

Add variable-data directories, return the number of directories added.

load_variables(var_loads: LoadedFiles = (None, None), name: str = '__root__') → LoadResult

Load variable data, resolve any un-loaded variable directories.

Module contents

datazen - A data structure for runtime data.

class datazen.environment.EnvironmentNamespace(name: str)

Bases: object

The base class for management of the environment data structure.

add_dir(dir_type: DataType, dir_path: str, rel_path: str = '.', allow_dup: bool = False) → bool

Add a directory to be loaded for a given data type.

add_dirs(dir_type: DataType, dir_paths: List[str], rel_path: str = '.', allow_dup: bool = False) → int

Add multiple directories for a given data type, return the number of directories added.

get_to_load(dir_type: DataType) → List[Path | str | None]

Build a list of the yet-to-be-loaded directories for a given data type.

unload(dir_type: DataType) → None

Mark all directories for a given type as un-loaded.

unload_all() → None

Mark all directories as unloaded.

update_load_state(dir_type: DataType, to_load: List[Path | str | None]) → int

Update the load states of directories in ‘to_load’ for a given data type.

datazen.environment.clone(env: EnvironmentNamespace, update: EnvironmentNamespace) → EnvironmentNamespace

Create a clone (deep copy) of an existing Environment.