bedrock.config.models

 1import sys  # pragma: unit
 2from os.path import basename  # pragma: unit
 3import os  # pragma: unit
 4from bedrock._helpers.string import snake_case_to_camelCase  # pragma: unit
 5from bedrock.log import log_config  # pragma: unit
 6from bedrock.config._common import _get_module_from_path  # pragma: unit
 7
 8log = log_config("models_config")  # pragma: unit
 9
10
11def get_models(models_path="../model", module_prefix="model") -> list:  # pragma: unit
12    """
13    Dynamically imports and finds all the models from the provided model directory.
14    This is primarily for initialising models for SQLAlchemy and to generate the DB schema.
15
16    Usually an internal thing, but can be useful for the `/status/` endpoint.
17
18    :param models_path: The path to the models' directory. Usually `config.internals["models_location"]`
19    :param module_prefix: Prefix to add to the module
20    :return: A list of all the defined models.
21    """
22    return [*get_models_dictionary(models_path, module_prefix).values()]
23
24
25def get_models_dictionary(models_path: str = "../model", module_prefix: str = "model",
26                          module_path: str or None = None) -> dict:  # pragma: unit
27    """
28    Dynamically imports and finds all the models from the provided model directory.
29
30    :param models_path: The path to the models' directory. Usually `config.internals["models_location"]`
31    :param module_prefix: Prefix to add to the module
32    :param module_path: The path to the module. If None, it will be the same as models_path.
33    :return: A list of all the defined models.
34    """
35    log.debug(f"Loading models from {models_path}...")
36    directory = os.fsencode(models_path)
37    excludes = []
38    modules_list = [basename(f)[:-3].decode("utf-8") for f in os.listdir(directory) if
39               f not in excludes
40               and not f.startswith(b"__")]
41
42    log.debug(f"Found modules {modules_list}")
43    models = {}
44    for module_name in modules_list:
45        loaded_modules = sys.modules
46        class_name = snake_case_to_camelCase(module_name, True)
47        full_module_name = f'{module_prefix}.{module_name}'
48        if module_path is None or ".py" in module_name:
49            _module_path = module_path
50        else:
51            _module_path = f"{models_path}/{module_name}.py"
52        if full_module_name not in loaded_modules.keys():
53            _get_module_from_path(full_module_name, _module_path)
54        model = getattr(loaded_modules[full_module_name], class_name)
55        models[class_name] = model
56    return models