lakehouse_engine.utils.acon_utils

 1"""Module to perform validations and resolve the acon."""
 2
 3from lakehouse_engine.core.definitions import DQType, InputFormat, OutputFormat
 4from lakehouse_engine.io.exceptions import WrongIOFormatException
 5from lakehouse_engine.utils.dq_utils import PrismaUtils
 6from lakehouse_engine.utils.logging_handler import LoggingHandler
 7
 8_LOGGER = LoggingHandler(__name__).get_logger()
 9
10
11def validate_and_resolve_acon(acon: dict, execution_point: str = "") -> dict:
12    """Function to validate and resolve the acon.
13
14    Args:
15        acon: Acon to be validated and resolved.
16        execution_point: Execution point to resolve the dq functions.
17
18    Returns:
19        Acon after validation and resolution.
20    """
21    # Performing validations
22    validate_readers(acon)
23    validate_writers(acon)
24
25    # Resolving the acon
26    if execution_point:
27        acon = resolve_dq_functions(acon, execution_point)
28
29    _LOGGER.info(f"Read Algorithm Configuration: {str(acon)}")
30
31    return acon
32
33
34def validate_readers(acon: dict) -> None:
35    """Function to validate the readers in the acon.
36
37    Args:
38        acon: Acon to be validated.
39
40    Raises:
41        RuntimeError: If the input format is not supported.
42    """
43    if "input_specs" in acon.keys() or "input_spec" in acon.keys():
44        for spec in acon.get("input_specs", []) or [acon.get("input_spec", {})]:
45            if (
46                not InputFormat.exists(spec.get("data_format"))
47                and "db_table" not in spec.keys()
48            ):
49                raise WrongIOFormatException(
50                    f"Input format not supported: {spec.get('data_format')}"
51                )
52
53
54def validate_writers(acon: dict) -> None:
55    """Function to validate the writers in the acon.
56
57    Args:
58        acon: Acon to be validated.
59
60    Raises:
61        RuntimeError: If the output format is not supported.
62    """
63    if "output_specs" in acon.keys() or "output_spec" in acon.keys():
64        for spec in acon.get("output_specs", []) or [acon.get("output_spec", {})]:
65            if not OutputFormat.exists(spec.get("data_format")):
66                raise WrongIOFormatException(
67                    f"Output format not supported: {spec.get('data_format')}"
68                )
69
70
71def resolve_dq_functions(acon: dict, execution_point: str) -> dict:
72    """Function to resolve the dq functions in the acon.
73
74    Args:
75        acon: Acon to resolve the dq functions.
76        execution_point: Execution point of the dq_functions.
77
78    Returns:
79        Acon after resolving the dq functions.
80    """
81    if acon.get("dq_spec"):
82        if acon.get("dq_spec").get("dq_type") == DQType.PRISMA.value:
83            acon["dq_spec"] = PrismaUtils.build_prisma_dq_spec(
84                spec=acon.get("dq_spec"), execution_point=execution_point
85            )
86    elif acon.get("dq_specs"):
87        resolved_dq_specs = []
88        for spec in acon.get("dq_specs", []):
89            if spec.get("dq_type") == DQType.PRISMA.value:
90                resolved_dq_specs.append(
91                    PrismaUtils.build_prisma_dq_spec(
92                        spec=spec, execution_point=execution_point
93                    )
94                )
95            else:
96                resolved_dq_specs.append(spec)
97        acon["dq_specs"] = resolved_dq_specs
98    return acon