-
Notifications
You must be signed in to change notification settings - Fork 1.5k
/
expectation_configuration_builder.py
140 lines (123 loc) · 5.31 KB
/
expectation_configuration_builder.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
import logging
from abc import ABC, abstractmethod
from typing import Dict, List, Optional, Set, Union
from great_expectations.core.batch import Batch, BatchRequest, RuntimeBatchRequest
from great_expectations.core.expectation_configuration import ExpectationConfiguration
from great_expectations.rule_based_profiler.config import ParameterBuilderConfig
from great_expectations.rule_based_profiler.helpers.util import (
init_rule_parameter_builders,
set_batch_list_or_batch_request_on_builder,
)
from great_expectations.rule_based_profiler.parameter_builder import ParameterBuilder
from great_expectations.rule_based_profiler.types import (
Builder,
Domain,
ParameterContainer,
)
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
class ExpectationConfigurationBuilder(Builder, ABC):
exclude_field_names: Set[str] = Builder.exclude_field_names | {
"validation_parameter_builders",
}
def __init__(
self,
expectation_type: str,
validation_parameter_builder_configs: Optional[
List[ParameterBuilderConfig]
] = None,
batch_list: Optional[List[Batch]] = None,
batch_request: Optional[
Union[str, BatchRequest, RuntimeBatchRequest, dict]
] = None,
data_context: Optional["DataContext"] = None, # noqa: F821
**kwargs
):
"""
The ExpectationConfigurationBuilder will build ExpectationConfiguration objects for a Domain from the Rule.
Args:
expectation_type: the "expectation_type" argument of "ExpectationConfiguration" object to be emitted.
validation_parameter_builder_configs: ParameterBuilder configurations, having whose outputs available (as
fully-qualified parameter names) is pre-requisite for present ExpectationConfigurationBuilder instance.
batch_list: explicitly passed Batch objects for parameter computation (take precedence over batch_request).
batch_request: specified in ParameterBuilder configuration to get Batch objects for parameter computation.
data_context: DataContext
kwargs: additional arguments
"""
super().__init__(
batch_list=batch_list,
batch_request=batch_request,
data_context=data_context,
)
self._expectation_type = expectation_type
self._validation_parameter_builders = init_rule_parameter_builders(
parameter_builder_configs=validation_parameter_builder_configs,
data_context=self._data_context,
)
"""
Since ExpectationConfigurationBuilderConfigSchema allows arbitrary fields (as ExpectationConfiguration kwargs)
to be provided, they must be all converted to public property accessors and/or public fields in order for all
provisions by Builder, SerializableDictDot, and DictDot to operate properly in compliance with their interfaces.
"""
for k, v in kwargs.items():
setattr(self, k, v)
logger.debug(
'Setting unknown kwarg (%s, %s) provided to constructor as argument in "%s".',
k,
v,
self.__class__.__name__,
)
def build_expectation_configuration(
self,
parameter_container: ParameterContainer,
domain: Domain,
variables: Optional[ParameterContainer] = None,
parameters: Optional[Dict[str, ParameterContainer]] = None,
) -> ExpectationConfiguration:
self._resolve_validation_dependencies(
parameter_container=parameter_container,
domain=domain,
variables=variables,
parameters=parameters,
)
return self._build_expectation_configuration(
domain=domain, variables=variables, parameters=parameters
)
def _resolve_validation_dependencies(
self,
parameter_container: ParameterContainer,
domain: Domain,
variables: Optional[ParameterContainer] = None,
parameters: Optional[Dict[str, ParameterContainer]] = None,
) -> None:
validation_parameter_builders: List[ParameterBuilder] = (
self.validation_parameter_builders or []
)
validation_parameter_builder: ParameterBuilder
for validation_parameter_builder in validation_parameter_builders:
set_batch_list_or_batch_request_on_builder(
builder=validation_parameter_builder,
batch_list=self.batch_list,
batch_request=self.batch_request,
force_batch_data=False,
)
validation_parameter_builder.build_parameters(
parameter_container=parameter_container,
domain=domain,
variables=variables,
parameters=parameters,
)
@abstractmethod
def _build_expectation_configuration(
self,
domain: Domain,
variables: Optional[ParameterContainer] = None,
parameters: Optional[Dict[str, ParameterContainer]] = None,
) -> ExpectationConfiguration:
pass
@property
def expectation_type(self) -> str:
return self._expectation_type
@property
def validation_parameter_builders(self) -> Optional[List[ParameterBuilder]]:
return self._validation_parameter_builders