datazen.environment.integrated
index
/home/vkottler/src/vkottler/workspace/datazen/datazen/environment/integrated.py

datazen - A centralized store for runtime data.

 
Modules
       
logging
os

 
Classes
       
datazen.environment.command.CommandEnvironment(datazen.environment.task.TaskEnvironment)
Environment(datazen.environment.compile.CompileEnvironment, datazen.environment.render.RenderEnvironment, datazen.environment.group.GroupEnvironment, datazen.environment.command.CommandEnvironment)
datazen.environment.compile.CompileEnvironment(datazen.environment.task.TaskEnvironment)
Environment(datazen.environment.compile.CompileEnvironment, datazen.environment.render.RenderEnvironment, datazen.environment.group.GroupEnvironment, datazen.environment.command.CommandEnvironment)
datazen.environment.group.GroupEnvironment(datazen.environment.task.TaskEnvironment)
Environment(datazen.environment.compile.CompileEnvironment, datazen.environment.render.RenderEnvironment, datazen.environment.group.GroupEnvironment, datazen.environment.command.CommandEnvironment)
datazen.environment.render.RenderEnvironment(datazen.environment.task.TaskEnvironment)
Environment(datazen.environment.compile.CompileEnvironment, datazen.environment.render.RenderEnvironment, datazen.environment.group.GroupEnvironment, datazen.environment.command.CommandEnvironment)

 
class Environment(datazen.environment.compile.CompileEnvironment, datazen.environment.render.RenderEnvironment, datazen.environment.group.GroupEnvironment, datazen.environment.command.CommandEnvironment)
    Environment(newline: str = '\n') -> None
 
A wrapper for inheriting all environment-loading capabilities.
 
 
Method resolution order:
Environment
datazen.environment.compile.CompileEnvironment
datazen.environment.render.RenderEnvironment
datazen.environment.group.GroupEnvironment
datazen.environment.command.CommandEnvironment
datazen.environment.task.TaskEnvironment
datazen.environment.manifest_cache.ManifestCacheEnvironment
datazen.environment.manifest.ManifestEnvironment
datazen.environment.config.ConfigEnvironment
datazen.environment.variable.VariableEnvironment
datazen.environment.schema.SchemaEnvironment
datazen.environment.template.TemplateEnvironment
datazen.environment.base.BaseEnvironment
builtins.object

Methods defined here:
__init__(self, newline: str = '\n') -> None
Add a notion of 'visited' targets to the environment data.
command(self, target: str) -> datazen.environment.base.TaskResult
Execute a named 'command' target from the manifest.
compile(self, target: str) -> datazen.environment.base.TaskResult
Execute a named 'compile' target from the manifest.
execute(self, target: str = '', should_cache: bool = True) -> datazen.environment.base.TaskResult
Execute an arbitrary target.
execute_targets(self, targets: List[str]) -> bool
Execute a list of named targets and return whether or not the entire
set was successful.
group(self, target: str) -> datazen.environment.base.TaskResult
Attempt to satisfy a 'group' target.
render(self, target: str) -> datazen.environment.base.TaskResult
Execute a named 'render' target from the manifest.

Methods inherited from datazen.environment.compile.CompileEnvironment:
valid_compile(self, entry: Dict[str, Any], namespace: str, dep_data: Dict[str, Any] = None, deps_changed: List[str] = None, logger: logging.Logger = <Logger datazen.environment.compile (WARNING)>) -> datazen.environment.base.TaskResult
Perform the compilation specified by the entry.

Methods inherited from datazen.environment.render.RenderEnvironment:
perform_render(self, template: jinja2.environment.Template, all_templates: Dict[str, jinja2.environment.Template], path: Optional[str], entry: Dict[str, Any], data: Dict[str, Any] = None, logger: logging.Logger = <Logger datazen.environment.render (WARNING)>) -> datazen.environment.base.TaskResult
Render a template to the requested path using the provided data.
store_render(self, entry: Dict[str, Any], data: Dict[str, Any]) -> None
Store data from the current render.
valid_render(self, entry: Dict[str, Any], namespace: str, dep_data: Dict[str, Any] = None, deps_changed: List[str] = None, logger: logging.Logger = <Logger datazen.environment.render (WARNING)>) -> datazen.environment.base.TaskResult
Perform the render specified by the entry.

Methods inherited from datazen.environment.group.GroupEnvironment:
valid_group(self, entry: Dict[str, Any], _: str, dep_data: Dict[str, Any] = None, deps_changed: List[str] = None, logger: logging.Logger = <Logger datazen.environment.group (WARNING)>) -> datazen.environment.base.TaskResult
Stub task to group other tasks.

Methods inherited from datazen.environment.command.CommandEnvironment:
valid_command(self, entry: Dict[str, Any], _: str, __: Dict[str, Any] = None, deps_changed: List[str] = None, logger: logging.Logger = <Logger datazen.environment.command (WARNING)>) -> datazen.environment.base.TaskResult
Perform the command specified by the entry.

Methods inherited from datazen.environment.task.TaskEnvironment:
already_satisfied(self, target: str, output_path: Optional[str], load_deps: List[str], deps_changed: List[str] = None, load_checks: Dict[str, 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(self, purge_data: bool = True) -> None
Remove cached data from the file-system.
get_dep_data(self, dep_list: 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(self, category: str, name: str) -> Dict[str, Any]
Get an entry from the manifest.
handle_task(self, key_name: str, target: str, task_stack: List[datazen.environment.base.Task] = None, should_cache: bool = True) -> datazen.environment.base.TaskResult
Handle the setup for manifest tasks, such as loading additional data
directories and setting the correct output directory.
init_cache(self, cache_dir: str) -> None
Initialize the task-data cache.
is_resolved(self, operation: str, target: str) -> bool
Determine whether a target for an operation has already been resolved.
is_task_new(self, operation: str, target: str) -> bool
Determine if a target has been newly executed this iteration.
push_dep(self, dep: str, task_stack: List[datazen.environment.base.Task], curr_target: str) -> None
Push a dependency onto a stack if they need to be resolved.
resolve(self, operation: str, target: str, should_cache: bool, is_new: bool) -> None
Set a target for an operation as resolved.
resolve_dependencies(self, dep_list: List[str], task_stack: 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.
valid_noop(self, entry: Dict[str, Any], _: str, __: Dict[str, Any] = None, ___: List[str] = None, logger: logging.Logger = <Logger datazen.environment.task (WARNING)>) -> datazen.environment.base.TaskResult
Default handle for a potentially-unregistered operation.
write_cache(self) -> None
Commit cached data to the file-system.

Readonly properties inherited from datazen.environment.task.TaskEnvironment:
task_data
Proxy task data through the cache.

Methods inherited from datazen.environment.manifest_cache.ManifestCacheEnvironment:
cached_enforce_schemas(self, data: Dict[str, Any], require_all: bool = True, name: str = '__root__') -> bool
Enforce schemas, proxied through the cache.
cached_load_configs(self, name: str = '__root__', enforce_schemas: bool = True) -> vcorelib.io.types.LoadResult
Load configs, proxied through the cache.
cached_load_schemas(self, require_all: bool = True, name: str = '__root__') -> Dict[str, Any]
Load schemas, proxied through the cache.
cached_load_templates(self, name: str = '__root__') -> Dict[str, jinja2.environment.Template]
Load templates, proxied through the cache.
cached_load_variables(self, name: str = '__root__') -> vcorelib.io.types.LoadResult
Load variables, proxied through the cache.
describe_cache(self) -> None
Describe the [initial] cache for debugging purposes.
get_new_loaded(self, 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(self, 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(self) -> None
Return the cache to its initially-loaded state.

Methods inherited from datazen.environment.manifest.ManifestEnvironment:
default_target(self) -> str
Retrieve a configured default target for a manifest, otherwise an
empty String.
load_dirs(self, 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(self, path: str = 'manifest.yaml', logger: logging.Logger = <Logger datazen.environment.manifest (WARNING)>) -> bool
Attempt to load manifest data from a file.
load_manifest_reent(self, path: str, manifest_dir: str, params: Dict[Any, Any], files: 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.

Static methods inherited from datazen.environment.manifest.ManifestEnvironment:
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).

Data and other attributes inherited from datazen.environment.manifest.ManifestEnvironment:
path_fields = ['configs', 'schemas', 'schema_types', 'variables', 'templates']
targets_with_paths = ['compiles', 'renders']

Methods inherited from datazen.environment.config.ConfigEnvironment:
add_config_dirs(self, 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(self, cfg_loads: datazen.load.LoadedFiles = LoadedFiles(files=None, file_data=None), var_loads: datazen.load.LoadedFiles = LoadedFiles(files=None, file_data=None), sch_loads: datazen.load.LoadedFiles = LoadedFiles(files=None, file_data=None), sch_types_loads: datazen.load.LoadedFiles = LoadedFiles(files=None, file_data=None), name: str = '__root__', logger: logging.Logger = <Logger datazen.environment.config (WARNING)>, enforce_schemas: bool = True) -> vcorelib.io.types.LoadResult
Load configuration data, resolve any un-loaded configuration
directories.

Methods inherited from datazen.environment.variable.VariableEnvironment:
add_variable_dirs(self, 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(self, var_loads: datazen.load.LoadedFiles = LoadedFiles(files=None, file_data=None), name: str = '__root__') -> vcorelib.io.types.LoadResult
Load variable data, resolve any un-loaded variable directories.

Methods inherited from datazen.environment.schema.SchemaEnvironment:
add_schema_dirs(self, 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(self, 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(self, data: Dict[str, Any], require_all: bool = True, sch_loads: datazen.load.LoadedFiles = LoadedFiles(files=None, file_data=None), sch_types_loads: datazen.load.LoadedFiles = LoadedFiles(files=None, file_data=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(self, sch_loads: datazen.load.LoadedFiles = LoadedFiles(files=None, file_data=None), name: str = '__root__') -> Dict[str, Any]
Load custom schema types and resolve any un-loaded directories.
load_schemas(self, require_all: bool = True, sch_loads: datazen.load.LoadedFiles = LoadedFiles(files=None, file_data=None), sch_types_loads: datazen.load.LoadedFiles = LoadedFiles(files=None, file_data=None), name: str = '__root__', modify_registry: bool = True) -> Dict[str, Any]
Load schema data, resolve any un-loaded schema directories.

Methods inherited from datazen.environment.template.TemplateEnvironment:
add_template_dirs(self, 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(self, template_loads: datazen.load.LoadedFiles = LoadedFiles(files=None, file_data=None), name: str = '__root__') -> Dict[str, jinja2.environment.Template]
Load templates, resolve any un-loaded template directories.

Methods inherited from datazen.environment.base.BaseEnvironment:
add_dir(self, dir_type: datazen.enums.DataType, dir_path: str, rel_path: str = '.', name: str = '__root__', allow_dup: bool = False) -> bool
Proxy for add_dir for a namespace.
add_dirs(self, dir_type: datazen.enums.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(self, name: str, clone_root: bool = True) -> None
Add a new namespace, optionally clone from the existing root.
get_data(self, dir_type: datazen.enums.DataType, name: str = '__root__') -> Dict[str, Any]
Get the raw data for a directory type from a namespace.
get_namespace(self, 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(self, dir_type: datazen.enums.DataType, name: str = '__root__') -> List[Union[pathlib.Path, str, NoneType]]
Proxy for a namespace's get_to_load.
get_valid(self, name: str = '__root__') -> bool
Get the 'valid' flag for a namespace.
set_valid(self, value: bool, name: str = '__root__') -> None
Set the 'valid' flag for a namespace.
unload_all(self, name: str = '__root__') -> None
Unload all of the directories for a namespace.
update_load_state(self, dir_type: datazen.enums.DataType, to_load: List[Union[pathlib.Path, str, NoneType]], name: str = '__root__') -> int
Proxy for update_load_state for a namespace.

Data descriptors inherited from datazen.environment.base.BaseEnvironment:
__dict__
dictionary for instance variables
__weakref__
list of weak references to the object

 
Functions
       
from_manifest(manifest_path: str, newline: str = '\n', data_cache_name: str = 'task_data', logger: logging.Logger = <Logger datazen.environment.integrated (WARNING)>) -> datazen.environment.integrated.Environment
Load an environment object from a schema definition on disk.

 
Data
        CACHE_SUFFIX = '_cache'
List = typing.List