pyfileconf package

py-file-conf is a Python framework for flow-based programming and managing configuration. To use it, you register your main functions and classes, and config files are created automatically for them. It provides a way to run these functions individually, in a list, in sections, or a combination thereof. Configuration can be dynamically updated, enabling powerful scripting.

Subpackages

Submodules

pyfileconf.batch module

class pyfileconf.batch.BatchUpdater(base_section_path_str=None, strip_manager_from_iv=False)[source]

Bases: object

Class that enables config changes across multiple pipeline managers

Update one or multiple registered functions/sections by their full section paths, regardless of in which manager they reside

__init__(base_section_path_str=None, strip_manager_from_iv=False)[source]
Parameters

  • base_section_path_str (Optional[str]) – section path str to put at beginning of all passed section paths

  • strip_manager_from_iv (bool) – whether to remove manager name from any incoming item views

reset(section_path_strs)[source]
strip_manager_from_iv: bool
update(updates)[source]
Parameters

updates (Iterable[Dict[str, Any]]) – list of kwarg dictionaries which would normally be provided to .update_batch

Return type

None

Returns

updates: Iterable[Dict[str, Any]]

pyfileconf.debug module

pyfileconf.debug.pdb_post_mortem_or_passed_debug_fn(*args, debug_fn=False, **kwargs)[source]
Return type

None

pyfileconf.debug.post_mortem(*args, debug_fn=None, **kwargs)[source]

Post-mortem, using a custom debug function if passed

Parameters

debug_fn (Optional[Callable]) –

Return type

None

Returns

pyfileconf.interfaces module

class pyfileconf.interfaces.SpecificClassConfigDict(**kwargs)

Bases: dict

always_assign_strs
always_import_strs
class
execute_attr
key_attr
name

pyfileconf.iterate module

class pyfileconf.iterate.IterativeRunner(section_path_str_or_list, config_updates, base_section_path_str=None, strip_manager_from_iv=False)[source]

Bases: object

Class that enables running functions/sections with different combinations of config updates, even across multiple pipeline managers

Run one or multiple registered functions/sections multiple times, each time updating the config with a combination of the passed config updates.

Aggregates the updates to each config, then takes the itertools product of all the config updates to run the functions with each combination of the configs.

__init__(section_path_str_or_list, config_updates, base_section_path_str=None, strip_manager_from_iv=False)[source]
Parameters

  • section_path_str_or_list (Union[str, List[str], ForwardRef, List[ForwardRef]]) – . separated name of path of function or section, or list thereof. similar to how a function would be imported. e.g. ‘main.data.summarize.summary_func1’ or when running multiple functions/sections, e.g. [‘main.data’, ‘main.analysis.reg.1’]

  • config_updates (Sequence[Dict[str, Any]]) – list of kwarg dictionaries which would normally be provided to .update

  • base_section_path_str (Optional[str]) – section path str to put at beginning of all passed section paths

  • strip_manager_from_iv (bool) – whether to remove manager name from any incoming item views

cases: List[Tuple[Dict[str, Any], …]]
config_updates: Sequence[Dict[str, Any]]
defaults: Dict[str, Dict[str, Any]]
get_cases()[source]
Return type

List[Tuple[Dict[str, Any], …]]

get_defaults()[source]
Return type

Dict[str, Dict[str, Any]]

last_case: Optional[Tuple[Dict[str, Any], …]] = None
run(collect_results=True)[source]
Parameters

collect_results (bool) – Whether to aggregate and return results, set to False to save memory if results are stored in some other way

Return type

List[Tuple[Tuple[Dict[str, Any], …], Tuple[Any]]]

Returns

run_gen()[source]
Return type

Iterator[Tuple[Tuple[Dict[str, Any], …], Tuple[Any]]]

run_items: List[SectionPath]
should_run(case)[source]

Hook which can be overriden in subclass to determine whether case should be run

The default hook always returns True, the case should be run.

Parameters

case (Tuple[Dict[str, Any], …]) – Tuple of config dicts being updated

Return type

bool

Returns

pyfileconf.iterate.get_config_product(config_dicts)[source]
Parameters

config_dicts (Sequence[Dict[str, Any]]) –

Return type

List[Tuple[Dict[str, Any], …]]

Returns

Examples
>>> cd = dict(
...     section_path_str='abc',
...     a=10
... )
>>> cd2 = dict(
...     section_path_str='abc',
...     a=20
... )
>>> cd3 = dict(
...     section_path_str='def',
...     a=20
... )
>>> cds = [cd, cd2, cd3]
>>> result = get_config_product(cds)
>>> result
... [({'section_path_str': 'abc', 'a': 10}, {'section_path_str': 'def', 'a': 20}),
... ({'section_path_str': 'abc', 'a': 20}, {'section_path_str': 'def', 'a': 20})]

pyfileconf.main module

class pyfileconf.main.PipelineManager(folder, name='project', specific_class_config_dicts=None, auto_pdb=False, force_continue=False, default_config_folder_name='defaults')[source]

Bases: object

Main class for managing flow-based programming and configuration.

__init__(folder, name='project', specific_class_config_dicts=None, auto_pdb=False, force_continue=False, default_config_folder_name='defaults')[source]

Initialize self. See help(type(self)) for accurate signature.

create(section_path_str, func_or_class=None)[source]

Create a new configuration entry dynamically rather than manually modifying dict file

Parameters

  • func_or_class (Union[Callable, Type, None]) – function or class to use to generate config

  • section_path_str (str) – section path at which the config should be stored

Returns

get(section_path_str_or_view)[source]
classmethod get_manager_by_filepath(filepath)[source]

Gets the active pipeline manager which references the passed filepath.

Parameters

filepath (str) –

Return type

PipelineManager

Returns

Raises

NoPipelineManagerForFilepathException – When no active pipeline manager has a config file matching the file path

classmethod get_manager_by_section_path_str(section_path_str)[source]

Gets the active pipeline manager which references the passed section path.

Parameters

filepath –

Return type

PipelineManager

Returns

Raises

NoPipelineManagerForSectionPathException – When no active pipeline manager has a config file matching the file path

load()[source]

Wrapper to track imported modules so that can reimport them upon reloading

Return type

None

refresh(section_path_str)[source]

Reloads from the existing file, then reapplies any config updates. Useful for when this config depends on the attribute of some other config which was updated.

Parameters

section_path_str (str) – section path of item to be refreshed

Returns

reload()[source]

Useful for getting file system changes without having to start a new Python session. Also resets any locally defined configuration.

Reloads functions from pipeline dict, scaffolds config files for any new functions, and updates configuration from files.

Returns: None

Return type

None

reset(section_path_str_or_view, allow_create=False)[source]

Resets a function or section config to default.

To reset all configs, use .reload() instead.

Parameters

section_path_str_or_view (StrOrView) –

Returns

run(section_path_str_or_list)[source]

Use to run registered pipelines/functions/sections. Pass a single section path or a list of section paths. If a list is passed, the return value will also be a list, with each result item corresponding to the function at the same index.

If a path for a section is passed rather than a path for a function or pipeline, a list of results will be returned for that section as well. Therefore, calling a high-level section will result in a nested list structure of results.

Args:
section_path_str_or_list: . separated name of path of function or section, or list thereof.

similar to how a function would be imported. e.g. ‘main.data.summarize.summary_func1’ or when running multiple functions/sections, e.g.

[‘main.data’, ‘main.analysis.reg.1’]

These sections/functions are based on the structure of your pipeline_dict

Returns: result or list of results

Return type

Union[Any, List[Any]]

run_product(section_path_str_or_list, config_updates, collect_results=True)[source]

Run one or multiple registered functions/sections multiple times, each time updating the config with a combination of the passed config updates.

Aggregates the updates to each config, then takes the itertools product of all the config updates to run the functions with each combination of the configs.

Parameters

  • section_path_str_or_list (RunnerArgs) – . separated name of path of function or section, or list thereof. similar to how a function would be imported. e.g. ‘main.data.summarize.summary_func1’ or when running multiple functions/sections, e.g. [‘main.data’, ‘main.analysis.reg.1’]

  • config_updates (Sequence[Dict[str, Any]]) – list of kwarg dictionaries which would normally be provided to .update

  • collect_results (bool) – Whether to aggregate and return results, set to False to save memory if results are stored in some other way

Return type

IterativeResults

Returns

run_product_gen(section_path_str_or_list, config_updates)[source]
Return type

Iterator[ForwardRef]

update(d_=None, section_path_str=None, pyfileconf_persist=True, **kwargs)[source]

Update the configuration for an item by section path.

The main logic is in _update but this also calls the pyfileconf_pre_update and pyfileconf_post_update hooks

Parameters

  • d_ – dictionary of updates

  • section_path_str (Optional[str]) – section path of item to be updated

  • pyfileconf_persist (bool) – whether to make changes last through refreshing config

  • kwargs – kwarg updates

Returns

update_batch(updates)[source]

Update the configuration for an multiple items by section path.

The main logic is just calling _update in a loop but this also calls the pyfileconf_pre_update_batch and pyfileconf_post_update_batch hooks

Parameters

updates (Iterable[dict]) – iterable of dictionaries of config updates

Returns

None

exception pyfileconf.main.RunnerException(*args, section_path_str=None, trace_back=None)[source]

Bases: Exception

__init__(*args, section_path_str=None, trace_back=None)[source]

Initialize self. See help(type(self)) for accurate signature.

pyfileconf.main.create_collections(specific_class_config_dicts, basepath, pipeline_folder, pipeline_dict_path, manager_name=None)[source]
Return type

Tuple[List[SpecificClassCollection], PipelineCollection]

pyfileconf.main.create_project(path, logs_path, specific_class_config_dicts=None)[source]

Creates a new pyfileconf project file structure

pyfileconf.main.create_registrars(specific_class_config_dicts, basepath, pipeline_folder, pipeline_dict_path, manager_name=None)[source]
Return type

Tuple[List[SpecificRegistrar], PipelineRegistrar]

pyfileconf.main.report_runner_exceptions(runner_exceptions)[source]
Return type

None

pyfileconf.userbase module

class pyfileconf.userbase.PyfileconfBase[source]

Bases: object

A base class which can be used to autocomplete the possible custom methods which modify how pyfileconf works with the object.

It is not necessary to use this base class, instead the methods can be defined on any class.