Skip to content

plpipes.init#

init(*configs, config_files=[]) #

Initializes the PLPipes configuration.

This function merges provided configurations, command line arguments, and environment variables to set up the application. It also initializes logging and file system paths.

Parameters:

Name Type Description Default
*configs

Additional configuration dictionaries.

 ()
config_files

List of custom configuration files to be merged.

 []

Returns:

Name Type Description
bool

True if initialization is successful.
Source code in src\plpipes\init.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def init(*configs, config_files=[]):
    """Initializes the PLPipes configuration.

    This function merges provided configurations, command line arguments, and environment variables
    to set up the application. It also initializes logging and file system paths.

    Args:
        *configs: Additional configuration dictionaries.
        config_files: List of custom configuration files to be merged.

    Returns:
        bool: True if initialization is successful.
    """
    from pathlib import Path

    global _initialized
    if _initialized:
        logging.warning("Reinitialicing PLPipes")

    # frame 0: command line arguments
    # frame 1: custom configuration files
    # frame 2: standard configuration files

    cfg.merge(_config0, frame=2)

    for config in configs:
        for k, v in config.items():
            cfg.merge(v, key=k, frame=0)

    prog = Path(sys.argv[0])
    default_stem = str(prog.stem)

    # update cfg to get the log levels (normal and file)
    for fn in config_files:
        cfg.merge_file(fn, frame=1)

    list_configuration_files_not_found = []

    global_cfg_dir = Path.home() / ".config/plpipes"
    for suffix in ("", "-secrets"):
        for ext in ("json", "yml", "yaml"):
            path = global_cfg_dir / f"plpipes{suffix}.{ext}"
            if path.exists():
                cfg.merge_file(path, frame=2)
            else:
                list_configuration_files_not_found.append(path)

    for dir_key in (False, True):
        for stem_key in (False, True):
            for secrets_part in ("", "-secrets"):
                for env_key in (False, True):
                    for ext in ("json", "yml", "yaml"):
                        # The following values can be changed as
                        # config files are read, so they are
                        # recalculated every time:

                        env         = cfg.get('env', 'dev')
                        stem        = cfg.get('fs.stem', default_stem)
                        root_dir    = Path(cfg.get('fs.root'   , prog.parent.parent.absolute()))
                        config_dir  = Path(cfg.get('fs.config' , root_dir / "config"))
                        default_dir = Path(cfg.get('fs.default', root_dir / "default"))

                        env_part  = f"-{env}"  if env_key  else ""
                        stem_part = stem       if stem_key else "common"
                        dir       = config_dir if dir_key  else default_dir
                        path      = dir / f"{stem_part}{secrets_part}{env_part}.{ext}"
                        if path.exists():
                            cfg.merge_file(path, frame=2)
                        else:
                            list_configuration_files_not_found.append(path)

    # set the root log level as NOTSET, which is the deepest level; it's like
    # this because all the other handlers, even if they have ther own levels,
    # will not log anything if the root level is higher than their level
    logging.getLogger().setLevel("NOTSET")

    cfg.squash_frames()

    cfg.setdefault('fs.stem', default_stem)

    # calculate configuration for file system paths and set it
    root_dir = Path(cfg.setdefault('fs.root', prog.parent.parent.absolute()))
    for e in ('bin', 'lib', 'config', 'default',
              'input', 'output', 'work', 'actions',
              'resources'):
        cfg.setdefault("fs." + e, root_dir / e)

    init_run_as_of_date()

    _log_setup()

    logging.debug(f"List of configuration files not found: {list_configuration_files_not_found}")

    logging.debug(f"Configuration: {repr(cfg.to_tree())}")

    _initialized = True

    return True

init_run_as_of_date() #

Initializes the 'as_of_date' configuration entry.

It sets the 'run.as_of_date_normalized' entry in the configuration based on the specified or default 'run.as_of_date'.

Returns:

Type Description

None
Source code in src\plpipes\init.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
def init_run_as_of_date():
    """Initializes the 'as_of_date' configuration entry.

    It sets the 'run.as_of_date_normalized' entry in the configuration
    based on the specified or default 'run.as_of_date'.

    Returns:
        None
    """
    date = cfg.setdefault('run.as_of_date', 'now')
    as_of_date = friendlydateparser.parse_datetime(date)
    as_of_date = as_of_date.astimezone(datetime.timezone.utc)
    cfg['run.as_of_date_normalized'] = as_of_date.strftime("%Y%m%dT%H%M%SZ0")