bedrock.config.websockets

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