20220908.21

Build # 20220908.21

Build Type

push

circleci

Committed by GitHub

Commit Message

<a href="https://github.com/great-expectations/great_expectations/commit/<a class=hub.com/great-expectations/great_expectations/commit/8991338ad6260aa439c5c9588ec2f1bebce472c3">8991338ad<a href="https://github.com/great-expectations/great_expectations/commit/8991338ad6260aa439c5c9588ec2f1bebce472c3">">Merge </a><a class="double-link" href="https://github.com/great-expectations/great_expectations/commit/<a class="double-link" href="https://github.com/great-expectations/great_expectations/commit/0363b9e7f387dc5a5e5d2619e9375b893fccb580">0363b9e7f</a>">0363b9e7f</a><a href="https://github.com/great-expectations/great_expectations/commit/8991338ad6260aa439c5c9588ec2f1bebce472c3"> into cbc58f1c1">cbc58f1c1</a>

Run Details

19277 of 49524 relevant lines covered (38.92%)

1.14 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

11.36

/great_expectations/checkpoint/util.py

import copy
import json
import logging
import smtplib
import ssl
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from typing import List, Optional, Union

import requests

import great_expectations.exceptions as ge_exceptions
from great_expectations.core.batch import (
    BatchRequest,
    BatchRequestBase,
    RuntimeBatchRequest,
    get_batch_request_as_dict,
    materialize_batch_request,
)
from great_expectations.core.util import nested_update
from great_expectations.types import DictDot

try:
    import boto3
except ImportError:
    boto3 = None

logger = logging.getLogger(__name__)


def send_slack_notification(
    query, slack_webhook=None, slack_channel=None, slack_token=None
):
    session = requests.Session()
    url = slack_webhook
    query = query
    headers = None

    if not slack_webhook:
        url = "https://slack.com/api/chat.postMessage"
        headers = {"Authorization": f"Bearer {slack_token}"}
        query["channel"] = slack_channel

    try:
        response = session.post(url=url, headers=headers, json=query)
        if slack_webhook:
            ok_status = response.text == "ok"
        else:
            ok_status = response.json()["ok"]
    except requests.ConnectionError:
        logger.warning(f"Failed to connect to Slack webhook after {10} retries.")
    except Exception as e:
        logger.error(str(e))
    else:
        if response.status_code != 200 or not ok_status:
            logger.warning(
                "Request to Slack webhook "
                f"returned error {response.status_code}: {response.text}"
            )

        else:
            return "Slack notification succeeded."


# noinspection SpellCheckingInspection
def send_opsgenie_alert(query, suite_name, settings):
    """Creates an alert in Opsgenie."""
    if settings["region"] is not None:
        url = f"https://api.{settings['region']}.opsgenie.com/v2/alerts"  # accommodate for Europeans
    else:
        url = "https://api.opsgenie.com/v2/alerts"

    headers = {"Authorization": f"GenieKey {settings['api_key']}"}
    payload = {
        "message": f"Great Expectations suite {suite_name} failed",
        "description": query,
        "priority": settings["priority"],  # allow this to be modified in settings
        "tags": settings["tags"],
    }

    session = requests.Session()

    try:
        response = session.post(url, headers=headers, json=payload)
    except requests.ConnectionError:
        logger.warning("Failed to connect to Opsgenie")
    except Exception as e:
        logger.error(str(e))
    else:
        if response.status_code != 202:
            logger.warning(
                "Request to Opsgenie API "
                f"returned error {response.status_code}: {response.text}"
            )
        else:
            return "success"
    return "error"


def send_microsoft_teams_notifications(query, microsoft_teams_webhook):
    session = requests.Session()
    try:
        response = session.post(url=microsoft_teams_webhook, json=query)
    except requests.ConnectionError:
        logger.warning("Failed to connect to Microsoft Teams webhook after 10 retries.")

    except Exception as e:
        logger.error(str(e))
    else:
        if response.status_code != 200:
            logger.warning(
                "Request to Microsoft Teams webhook "
                f"returned error {response.status_code}: {response.text}"
            )
            return
        else:
            return "Microsoft Teams notification succeeded."


def send_webhook_notifications(query, webhook, target_platform):
    session = requests.Session()
    try:
        response = session.post(url=webhook, json=query)
    except requests.ConnectionError:
        logger.warning(
            f"Failed to connect to {target_platform} webhook after 10 retries."
        )
    except Exception as e:
        logger.error(str(e))
    else:
        if response.status_code != 200:
            logger.warning(
                f"Request to {target_platform} webhook "
                f"returned error {response.status_code}: {response.text}"
            )
        else:
            return f"{target_platform} notification succeeded."


# noinspection SpellCheckingInspection
def send_email(
    title,
    html,
    smtp_address,
    smtp_port,
    sender_login,
    sender_password,
    sender_alias,
    receiver_emails_list,
    use_tls,
    use_ssl,
):
    msg = MIMEMultipart()
    msg["From"] = sender_alias
    msg["To"] = ", ".join(receiver_emails_list)
    msg["Subject"] = title
    msg.attach(MIMEText(html, "html"))
    try:
        if use_ssl:
            if use_tls:
                logger.warning("Please choose between SSL or TLS, will default to SSL")
            context = ssl.create_default_context()
            mailserver = smtplib.SMTP_SSL(smtp_address, smtp_port, context=context)
        elif use_tls:
            mailserver = smtplib.SMTP(smtp_address, smtp_port)
            context = ssl.create_default_context()
            mailserver.starttls(context=context)
        else:
            logger.warning("Not using TLS or SSL to send an email is not secure")
            mailserver = smtplib.SMTP(smtp_address, smtp_port)
        mailserver.login(sender_login, sender_password)
        mailserver.sendmail(sender_alias, receiver_emails_list, msg.as_string())
        mailserver.quit()
    except smtplib.SMTPConnectError:
        logger.error(f"Failed to connect to the SMTP server at address: {smtp_address}")
    except smtplib.SMTPAuthenticationError:
        logger.error(
            f"Failed to authenticate to the SMTP server at address: {smtp_address}"
        )
    except Exception as e:
        logger.error(str(e))
    else:
        return "success"


def get_substituted_validation_dict(
    substituted_runtime_config: dict, validation_dict: dict
) -> dict:
    substituted_validation_dict = {
        "batch_request": get_substituted_batch_request(
            substituted_runtime_config=substituted_runtime_config,
            validation_batch_request=validation_dict.get("batch_request"),
        ),
        "expectation_suite_name": validation_dict.get("expectation_suite_name")
        or substituted_runtime_config.get("expectation_suite_name"),
        "expectation_suite_ge_cloud_id": validation_dict.get(
            "expectation_suite_ge_cloud_id"
        )
        or substituted_runtime_config.get("expectation_suite_ge_cloud_id"),
        "action_list": get_updated_action_list(
            base_action_list=substituted_runtime_config.get("action_list"),
            other_action_list=validation_dict.get("action_list", {}),
        ),
        "evaluation_parameters": nested_update(
            substituted_runtime_config.get("evaluation_parameters") or {},
            validation_dict.get("evaluation_parameters", {}),
            dedup=True,
        ),
        "runtime_configuration": nested_update(
            substituted_runtime_config.get("runtime_configuration") or {},
            validation_dict.get("runtime_configuration", {}),
            dedup=True,
        ),
        "include_rendered_content": validation_dict.get("include_rendered_content")
        or substituted_runtime_config.get("include_rendered_content")
        or None,
    }

    for attr in ("name", "id"):
        if validation_dict.get(attr) is not None:
            substituted_validation_dict[attr] = validation_dict[attr]

    validate_validation_dict(substituted_validation_dict)
    return substituted_validation_dict


# TODO: <Alex>A common utility function should be factored out from DataContext.get_batch_list() for any purpose.</Alex>
def get_substituted_batch_request(
    substituted_runtime_config: dict,
    validation_batch_request: Optional[Union[BatchRequestBase, dict]] = None,
) -> Optional[Union[BatchRequest, RuntimeBatchRequest]]:
    substituted_runtime_batch_request = substituted_runtime_config.get("batch_request")

    if substituted_runtime_batch_request is None and validation_batch_request is None:
        return None

    if substituted_runtime_batch_request is None:
        substituted_runtime_batch_request = {}

    if validation_batch_request is None:
        validation_batch_request = {}

    validation_batch_request = get_batch_request_as_dict(
        batch_request=validation_batch_request
    )
    substituted_runtime_batch_request = get_batch_request_as_dict(
        batch_request=substituted_runtime_batch_request
    )

    for key, value in validation_batch_request.items():
        substituted_value = substituted_runtime_batch_request.get(key)
        if value is not None and substituted_value is not None:
            raise ge_exceptions.CheckpointError(
                f'BatchRequest attribute "{key}" was specified in both validation and top-level CheckpointConfig.'
            )

    effective_batch_request: dict = dict(
        **substituted_runtime_batch_request, **validation_batch_request
    )

    return materialize_batch_request(batch_request=effective_batch_request)


def substitute_template_config(source_config: dict, template_config: dict) -> dict:
    if not (template_config and any(template_config.values())):
        return source_config

    dest_config: dict = copy.deepcopy(template_config)

    # replace
    if source_config.get("name") is not None:
        dest_config["name"] = source_config["name"]
    if source_config.get("module_name") is not None:
        dest_config["module_name"] = source_config["module_name"]
    if source_config.get("class_name") is not None:
        dest_config["class_name"] = source_config["class_name"]
    if source_config.get("run_name_template") is not None:
        dest_config["run_name_template"] = source_config["run_name_template"]
    if source_config.get("expectation_suite_name") is not None:
        dest_config["expectation_suite_name"] = source_config["expectation_suite_name"]
    if source_config.get("expectation_suite_ge_cloud_id") is not None:
        dest_config["expectation_suite_ge_cloud_id"] = source_config[
            "expectation_suite_ge_cloud_id"
        ]

    # update
    if source_config.get("batch_request") is not None:
        batch_request = dest_config.get("batch_request") or {}
        updated_batch_request = nested_update(
            batch_request,
            source_config["batch_request"],
            dedup=True,
        )
        dest_config["batch_request"] = updated_batch_request
    if source_config.get("action_list") is not None:
        action_list = dest_config.get("action_list") or []
        dest_config["action_list"] = get_updated_action_list(
            base_action_list=action_list,
            other_action_list=source_config["action_list"],
        )
    if source_config.get("evaluation_parameters") is not None:
        evaluation_parameters = dest_config.get("evaluation_parameters") or {}
        updated_evaluation_parameters = nested_update(
            evaluation_parameters,
            source_config["evaluation_parameters"],
            dedup=True,
        )
        dest_config["evaluation_parameters"] = updated_evaluation_parameters
    if source_config.get("runtime_configuration") is not None:
        runtime_configuration = dest_config.get("runtime_configuration") or {}
        updated_runtime_configuration = nested_update(
            runtime_configuration,
            source_config["runtime_configuration"],
            dedup=True,
        )
        dest_config["runtime_configuration"] = updated_runtime_configuration
    if source_config.get("validations") is not None:
        validations = dest_config.get("validations") or []
        existing_validations = template_config.get("validations") or []
        validations.extend(
            filter(
                lambda v: v not in existing_validations, source_config["validations"]
            )
        )
        dest_config["validations"] = validations
    if source_config.get("profilers") is not None:
        profilers = dest_config.get("profilers") or []
        existing_profilers = template_config.get("profilers") or []
        profilers.extend(
            filter(lambda v: v not in existing_profilers, source_config["profilers"])
        )
        dest_config["profilers"] = profilers

    return dest_config


def substitute_runtime_config(source_config: dict, runtime_kwargs: dict) -> dict:
    if not (runtime_kwargs and any(runtime_kwargs.values())):
        return source_config

    dest_config: dict = copy.deepcopy(source_config)

    # replace
    if runtime_kwargs.get("template_name") is not None:
        dest_config["template_name"] = runtime_kwargs["template_name"]
    if runtime_kwargs.get("run_name_template") is not None:
        dest_config["run_name_template"] = runtime_kwargs["run_name_template"]
    if runtime_kwargs.get("expectation_suite_name") is not None:
        dest_config["expectation_suite_name"] = runtime_kwargs["expectation_suite_name"]
    if runtime_kwargs.get("expectation_suite_ge_cloud_id") is not None:
        dest_config["expectation_suite_ge_cloud_id"] = runtime_kwargs[
            "expectation_suite_ge_cloud_id"
        ]
    # update
    if runtime_kwargs.get("batch_request") is not None:
        batch_request = dest_config.get("batch_request") or {}
        batch_request_from_runtime_kwargs = runtime_kwargs["batch_request"]
        batch_request_from_runtime_kwargs = get_batch_request_as_dict(
            batch_request=batch_request_from_runtime_kwargs
        )
        updated_batch_request = nested_update(
            batch_request,
            batch_request_from_runtime_kwargs,
            dedup=True,
        )
        dest_config["batch_request"] = updated_batch_request
    if runtime_kwargs.get("action_list") is not None:
        action_list = dest_config.get("action_list") or []
        dest_config["action_list"] = get_updated_action_list(
            base_action_list=action_list,
            other_action_list=runtime_kwargs["action_list"],
        )
    if runtime_kwargs.get("evaluation_parameters") is not None:
        evaluation_parameters = dest_config.get("evaluation_parameters") or {}
        updated_evaluation_parameters = nested_update(
            evaluation_parameters,
            runtime_kwargs["evaluation_parameters"],
            dedup=True,
        )
        dest_config["evaluation_parameters"] = updated_evaluation_parameters
    if runtime_kwargs.get("runtime_configuration") is not None:
        runtime_configuration = dest_config.get("runtime_configuration") or {}
        updated_runtime_configuration = nested_update(
            runtime_configuration,
            runtime_kwargs["runtime_configuration"],
            dedup=True,
        )
        dest_config["runtime_configuration"] = updated_runtime_configuration
    if runtime_kwargs.get("validations") is not None:
        validations = dest_config.get("validations") or []
        existing_validations = source_config.get("validations") or []
        validations.extend(
            filter(
                lambda v: v not in existing_validations,
                runtime_kwargs["validations"],
            )
        )
        dest_config["validations"] = validations
    if runtime_kwargs.get("profilers") is not None:
        profilers = dest_config.get("profilers") or []
        existing_profilers = source_config.get("profilers") or []
        profilers.extend(
            filter(lambda v: v not in existing_profilers, runtime_kwargs["profilers"])
        )
        dest_config["profilers"] = profilers

    return dest_config


def get_updated_action_list(
    base_action_list: list, other_action_list: list
) -> List[dict]:
    if base_action_list is None:
        base_action_list = []

    base_action_list_dict = {action["name"]: action for action in base_action_list}

    for other_action in other_action_list:
        other_action_name = other_action["name"]
        if other_action_name in base_action_list_dict:
            if other_action["action"]:
                nested_update(
                    base_action_list_dict[other_action_name],
                    other_action,
                    dedup=True,
                )
            else:
                base_action_list_dict.pop(other_action_name)
        else:
            base_action_list_dict[other_action_name] = other_action

    for other_action in other_action_list:
        other_action_name = other_action["name"]
        if other_action_name in base_action_list_dict:
            if not other_action["action"]:
                base_action_list_dict.pop(other_action_name)

    return list(base_action_list_dict.values())


def batch_request_in_validations_contains_batch_data(
    validations: Optional[List[dict]] = None,
) -> bool:
    if validations is not None:
        for idx, val in enumerate(validations):
            if (
                val.get("batch_request") is not None
                and isinstance(val.get("batch_request"), (dict, DictDot))
                and val["batch_request"].get("runtime_parameters") is not None
                and val["batch_request"]["runtime_parameters"].get("batch_data")
                is not None
            ):
                return True

    return False


def get_validations_with_batch_request_as_dict(
    validations: Optional[list] = None,
) -> Optional[list]:
    if validations:
        for value in validations:
            if "batch_request" in value:
                value["batch_request"] = get_batch_request_as_dict(
                    batch_request=value["batch_request"]
                )

    return validations


def validate_validation_dict(validation_dict: dict) -> None:
    if validation_dict.get("batch_request") is None:
        raise ge_exceptions.CheckpointError("validation batch_request cannot be None")
    if not validation_dict.get("expectation_suite_name"):
        raise ge_exceptions.CheckpointError(
            "validation expectation_suite_name must be specified"
        )
    if not validation_dict.get("action_list"):
        raise ge_exceptions.CheckpointError("validation action_list cannot be empty")


def send_cloud_notification(url: str, headers: dict):
    """
    Post a CloudNotificationAction to GE Cloud Backend for processing.
    """
    session = requests.Session()

    try:
        response = session.post(url=url, headers=headers)
    except requests.ConnectionError:
        logger.error(f"Failed to connect to Cloud backend after {10} retries.")
    except Exception as e:
        logger.error(str(e))
    else:
        if response.status_code != 200:
            message = f"Cloud Notification request returned error {response.status_code}: {response.text}"
            logger.error(message)
            return {"cloud_notification_result": message}
        else:
            return {"cloud_notification_result": "Cloud notification succeeded."}


def send_sns_notification(
    sns_topic_arn: str, sns_subject: str, validation_results: str, **kwargs
) -> str:
    """
    Send JSON results to an SNS topic with a schema of:


    :param sns_topic_arn:  The SNS Arn to publish messages to
    :param sns_subject: : The SNS Message Subject - defaults to expectation_suite_identifier.expectation_suite_name
    :param validation_results:  The results of the validation ran
    :param kwargs:  Keyword arguments to pass to the boto3 Session
    :return:  Message ID that was published

    """
    if not boto3:
        logger.warning("boto3 is not installed")
        return "boto3 is not installed"

    message_dict = {
        "TopicArn": sns_topic_arn,
        "Subject": sns_subject,
        "Message": json.dumps(validation_results),
        "MessageAttributes": {
            "String": {"DataType": "String.Array", "StringValue": "ValidationResults"},
        },
        "MessageStructure": "json",
    }
    session = boto3.Session(**kwargs)
    sns = session.client("sns")
    try:
        response = sns.publish(**message_dict)
    except sns.exceptions.InvalidParameterException:
        logger.error(f"Received invalid for message: {validation_results}")
    else:
        return f"Successfully posted results to {response['MessageId']} with Subject {sns_subject}"

1	import copy	3✔
2	import json	3✔
3	import logging	3✔
4	import smtplib	3✔
5	import ssl	3✔
6	from email.mime.multipart import MIMEMultipart	3✔
7	from email.mime.text import MIMEText	3✔
8	from typing import List, Optional, Union	3✔
9
10	import requests	3✔
11
12	import great_expectations.exceptions as ge_exceptions	3✔
13	from great_expectations.core.batch import (	3✔
14	BatchRequest,
15	BatchRequestBase,
16	RuntimeBatchRequest,
17	get_batch_request_as_dict,
18	materialize_batch_request,
19	)
20	from great_expectations.core.util import nested_update	3✔
21	from great_expectations.types import DictDot	3✔
22
23	try:	3✔
24	import boto3	3✔
25	except ImportError:	×
26	boto3 = None	×
27
28	logger = logging.getLogger(__name__)	3✔
29
30
31	def send_slack_notification(	3✔
32	query, slack_webhook=None, slack_channel=None, slack_token=None
33	):
34	session = requests.Session()	×
35	url = slack_webhook	×
36	query = query	×
37	headers = None	×
38
39	if not slack_webhook:	×
40	url = "https://slack.com/api/chat.postMessage"	×
41	headers = {"Authorization": f"Bearer {slack_token}"}	×
42	query["channel"] = slack_channel	×
43
44	try:	×
45	response = session.post(url=url, headers=headers, json=query)	×
46	if slack_webhook:	×
47	ok_status = response.text == "ok"	×
48	else:
49	ok_status = response.json()["ok"]	×
50	except requests.ConnectionError:	×
51	logger.warning(f"Failed to connect to Slack webhook after {10} retries.")	×
52	except Exception as e:	×
53	logger.error(str(e))	×
54	else:
55	if response.status_code != 200 or not ok_status:	×
56	logger.warning(	×
57	"Request to Slack webhook "
58	f"returned error {response.status_code}: {response.text}"
59	)
60
61	else:
62	return "Slack notification succeeded."	×
63
64
65	# noinspection SpellCheckingInspection
66	def send_opsgenie_alert(query, suite_name, settings):	3✔
67	"""Creates an alert in Opsgenie."""
68	if settings["region"] is not None:	×
69	url = f"https://api.{settings['region']}.opsgenie.com/v2/alerts" # accommodate for Europeans	×
70	else:
71	url = "https://api.opsgenie.com/v2/alerts"	×
72
73	headers = {"Authorization": f"GenieKey {settings['api_key']}"}	×
74	payload = {	×
75	"message": f"Great Expectations suite {suite_name} failed",
76	"description": query,
77	"priority": settings["priority"], # allow this to be modified in settings
78	"tags": settings["tags"],
79	}
80
81	session = requests.Session()	×
82
83	try:	×
84	response = session.post(url, headers=headers, json=payload)	×
85	except requests.ConnectionError:	×
86	logger.warning("Failed to connect to Opsgenie")	×
87	except Exception as e:	×
88	logger.error(str(e))	×
89	else:
90	if response.status_code != 202:	×
91	logger.warning(	×
92	"Request to Opsgenie API "
93	f"returned error {response.status_code}: {response.text}"
94	)
95	else:
96	return "success"	×
97	return "error"	×
98
99
100	def send_microsoft_teams_notifications(query, microsoft_teams_webhook):	3✔
101	session = requests.Session()	×
102	try:	×
103	response = session.post(url=microsoft_teams_webhook, json=query)	×
104	except requests.ConnectionError:	×
105	logger.warning("Failed to connect to Microsoft Teams webhook after 10 retries.")	×
106
107	except Exception as e:	×
108	logger.error(str(e))	×
109	else:
110	if response.status_code != 200:	×
111	logger.warning(	×
112	"Request to Microsoft Teams webhook "
113	f"returned error {response.status_code}: {response.text}"
114	)
115	return	×
116	else:
117	return "Microsoft Teams notification succeeded."	×
118
119
120	def send_webhook_notifications(query, webhook, target_platform):	3✔
121	session = requests.Session()	×
122	try:	×
123	response = session.post(url=webhook, json=query)	×
124	except requests.ConnectionError:	×
125	logger.warning(	×
126	f"Failed to connect to {target_platform} webhook after 10 retries."
127	)
128	except Exception as e:	×
129	logger.error(str(e))	×
130	else:
131	if response.status_code != 200:	×
132	logger.warning(	×
133	f"Request to {target_platform} webhook "
134	f"returned error {response.status_code}: {response.text}"
135	)
136	else:
137	return f"{target_platform} notification succeeded."	×
138
139
140	# noinspection SpellCheckingInspection
141	def send_email(	3✔
142	title,
143	html,
144	smtp_address,
145	smtp_port,
146	sender_login,
147	sender_password,
148	sender_alias,
149	receiver_emails_list,
150	use_tls,
151	use_ssl,
152	):
153	msg = MIMEMultipart()	×
154	msg["From"] = sender_alias	×
155	msg["To"] = ", ".join(receiver_emails_list)	×
156	msg["Subject"] = title	×
157	msg.attach(MIMEText(html, "html"))	×
158	try:	×
159	if use_ssl:	×
160	if use_tls:	×
161	logger.warning("Please choose between SSL or TLS, will default to SSL")	×
162	context = ssl.create_default_context()	×
163	mailserver = smtplib.SMTP_SSL(smtp_address, smtp_port, context=context)	×
164	elif use_tls:	×
165	mailserver = smtplib.SMTP(smtp_address, smtp_port)	×
166	context = ssl.create_default_context()	×
167	mailserver.starttls(context=context)	×
168	else:
169	logger.warning("Not using TLS or SSL to send an email is not secure")	×
170	mailserver = smtplib.SMTP(smtp_address, smtp_port)	×
171	mailserver.login(sender_login, sender_password)	×
172	mailserver.sendmail(sender_alias, receiver_emails_list, msg.as_string())	×
173	mailserver.quit()	×
174	except smtplib.SMTPConnectError:	×
175	logger.error(f"Failed to connect to the SMTP server at address: {smtp_address}")	×
176	except smtplib.SMTPAuthenticationError:	×
177	logger.error(	×
178	f"Failed to authenticate to the SMTP server at address: {smtp_address}"
179	)
180	except Exception as e:	×
181	logger.error(str(e))	×
182	else:
183	return "success"	×
184
185
186	def get_substituted_validation_dict(	3✔
187	substituted_runtime_config: dict, validation_dict: dict
188	) -> dict:
189	substituted_validation_dict = {	×
190	"batch_request": get_substituted_batch_request(
191	substituted_runtime_config=substituted_runtime_config,
192	validation_batch_request=validation_dict.get("batch_request"),
193	),
194	"expectation_suite_name": validation_dict.get("expectation_suite_name")
195	or substituted_runtime_config.get("expectation_suite_name"),
196	"expectation_suite_ge_cloud_id": validation_dict.get(
197	"expectation_suite_ge_cloud_id"
198	)
199	or substituted_runtime_config.get("expectation_suite_ge_cloud_id"),
200	"action_list": get_updated_action_list(
201	base_action_list=substituted_runtime_config.get("action_list"),
202	other_action_list=validation_dict.get("action_list", {}),
203	),
204	"evaluation_parameters": nested_update(
205	substituted_runtime_config.get("evaluation_parameters") or {},
206	validation_dict.get("evaluation_parameters", {}),
207	dedup=True,
208	),
209	"runtime_configuration": nested_update(
210	substituted_runtime_config.get("runtime_configuration") or {},
211	validation_dict.get("runtime_configuration", {}),
212	dedup=True,
213	),
214	"include_rendered_content": validation_dict.get("include_rendered_content")
215	or substituted_runtime_config.get("include_rendered_content")
216	or None,
217	}
218
219	for attr in ("name", "id"):	×
220	if validation_dict.get(attr) is not None:	×
221	substituted_validation_dict[attr] = validation_dict[attr]	×
222
223	validate_validation_dict(substituted_validation_dict)	×
224	return substituted_validation_dict	×
225
226
227	# TODO: <Alex>A common utility function should be factored out from DataContext.get_batch_list() for any purpose.</Alex>
228	def get_substituted_batch_request(	3✔
229	substituted_runtime_config: dict,
230	validation_batch_request: Optional[Union[BatchRequestBase, dict]] = None,
231	) -> Optional[Union[BatchRequest, RuntimeBatchRequest]]:
232	substituted_runtime_batch_request = substituted_runtime_config.get("batch_request")	×
233
234	if substituted_runtime_batch_request is None and validation_batch_request is None:	×
235	return None	×
236
237	if substituted_runtime_batch_request is None:	×
238	substituted_runtime_batch_request = {}	×
239
240	if validation_batch_request is None:	×
241	validation_batch_request = {}	×
242
243	validation_batch_request = get_batch_request_as_dict(	×
244	batch_request=validation_batch_request
245	)
246	substituted_runtime_batch_request = get_batch_request_as_dict(	×
247	batch_request=substituted_runtime_batch_request
248	)
249
250	for key, value in validation_batch_request.items():	×
251	substituted_value = substituted_runtime_batch_request.get(key)	×
252	if value is not None and substituted_value is not None:	×
253	raise ge_exceptions.CheckpointError(	×
254	f'BatchRequest attribute "{key}" was specified in both validation and top-level CheckpointConfig.'
255	)
256
257	effective_batch_request: dict = dict(	×
258	substituted_runtime_batch_request, validation_batch_request
259	)
260
261	return materialize_batch_request(batch_request=effective_batch_request)	×
262
263
264	def substitute_template_config(source_config: dict, template_config: dict) -> dict:	3✔
265	if not (template_config and any(template_config.values())):	×
266	return source_config	×
267
268	dest_config: dict = copy.deepcopy(template_config)	×
269
270	# replace
271	if source_config.get("name") is not None:	×
272	dest_config["name"] = source_config["name"]	×
273	if source_config.get("module_name") is not None:	×
274	dest_config["module_name"] = source_config["module_name"]	×
275	if source_config.get("class_name") is not None:	×
276	dest_config["class_name"] = source_config["class_name"]	×
277	if source_config.get("run_name_template") is not None:	×
278	dest_config["run_name_template"] = source_config["run_name_template"]	×
279	if source_config.get("expectation_suite_name") is not None:	×
280	dest_config["expectation_suite_name"] = source_config["expectation_suite_name"]	×
281	if source_config.get("expectation_suite_ge_cloud_id") is not None:	×
282	dest_config["expectation_suite_ge_cloud_id"] = source_config[	×
283	"expectation_suite_ge_cloud_id"
284	]
285
286	# update
287	if source_config.get("batch_request") is not None:	×
288	batch_request = dest_config.get("batch_request") or {}	×
289	updated_batch_request = nested_update(	×
290	batch_request,
291	source_config["batch_request"],
292	dedup=True,
293	)
294	dest_config["batch_request"] = updated_batch_request	×
295	if source_config.get("action_list") is not None:	×
296	action_list = dest_config.get("action_list") or []	×
297	dest_config["action_list"] = get_updated_action_list(	×
298	base_action_list=action_list,
299	other_action_list=source_config["action_list"],
300	)
301	if source_config.get("evaluation_parameters") is not None:	×
302	evaluation_parameters = dest_config.get("evaluation_parameters") or {}	×
303	updated_evaluation_parameters = nested_update(	×
304	evaluation_parameters,
305	source_config["evaluation_parameters"],
306	dedup=True,
307	)
308	dest_config["evaluation_parameters"] = updated_evaluation_parameters	×
309	if source_config.get("runtime_configuration") is not None:	×
310	runtime_configuration = dest_config.get("runtime_configuration") or {}	×
311	updated_runtime_configuration = nested_update(	×
312	runtime_configuration,
313	source_config["runtime_configuration"],
314	dedup=True,
315	)
316	dest_config["runtime_configuration"] = updated_runtime_configuration	×
317	if source_config.get("validations") is not None:	×
318	validations = dest_config.get("validations") or []	×
319	existing_validations = template_config.get("validations") or []	×
320	validations.extend(	×
321	filter(
322	lambda v: v not in existing_validations, source_config["validations"]
323	)
324	)
325	dest_config["validations"] = validations	×
326	if source_config.get("profilers") is not None:	×
327	profilers = dest_config.get("profilers") or []	×
328	existing_profilers = template_config.get("profilers") or []	×
329	profilers.extend(	×
330	filter(lambda v: v not in existing_profilers, source_config["profilers"])
331	)
332	dest_config["profilers"] = profilers	×
333
334	return dest_config	×
335
336
337	def substitute_runtime_config(source_config: dict, runtime_kwargs: dict) -> dict:	3✔
338	if not (runtime_kwargs and any(runtime_kwargs.values())):	×
339	return source_config	×
340
341	dest_config: dict = copy.deepcopy(source_config)	×
342
343	# replace
344	if runtime_kwargs.get("template_name") is not None:	×
345	dest_config["template_name"] = runtime_kwargs["template_name"]	×
346	if runtime_kwargs.get("run_name_template") is not None:	×
347	dest_config["run_name_template"] = runtime_kwargs["run_name_template"]	×
348	if runtime_kwargs.get("expectation_suite_name") is not None:	×
349	dest_config["expectation_suite_name"] = runtime_kwargs["expectation_suite_name"]	×
350	if runtime_kwargs.get("expectation_suite_ge_cloud_id") is not None:	×
351	dest_config["expectation_suite_ge_cloud_id"] = runtime_kwargs[	×
352	"expectation_suite_ge_cloud_id"
353	]
354	# update
355	if runtime_kwargs.get("batch_request") is not None:	×
356	batch_request = dest_config.get("batch_request") or {}	×
357	batch_request_from_runtime_kwargs = runtime_kwargs["batch_request"]	×
358	batch_request_from_runtime_kwargs = get_batch_request_as_dict(	×
359	batch_request=batch_request_from_runtime_kwargs
360	)
361	updated_batch_request = nested_update(	×
362	batch_request,
363	batch_request_from_runtime_kwargs,
364	dedup=True,
365	)
366	dest_config["batch_request"] = updated_batch_request	×
367	if runtime_kwargs.get("action_list") is not None:	×
368	action_list = dest_config.get("action_list") or []	×
369	dest_config["action_list"] = get_updated_action_list(	×
370	base_action_list=action_list,
371	other_action_list=runtime_kwargs["action_list"],
372	)
373	if runtime_kwargs.get("evaluation_parameters") is not None:	×
374	evaluation_parameters = dest_config.get("evaluation_parameters") or {}	×
375	updated_evaluation_parameters = nested_update(	×
376	evaluation_parameters,
377	runtime_kwargs["evaluation_parameters"],
378	dedup=True,
379	)
380	dest_config["evaluation_parameters"] = updated_evaluation_parameters	×
381	if runtime_kwargs.get("runtime_configuration") is not None:	×
382	runtime_configuration = dest_config.get("runtime_configuration") or {}	×
383	updated_runtime_configuration = nested_update(	×
384	runtime_configuration,
385	runtime_kwargs["runtime_configuration"],
386	dedup=True,
387	)
388	dest_config["runtime_configuration"] = updated_runtime_configuration	×
389	if runtime_kwargs.get("validations") is not None:	×
390	validations = dest_config.get("validations") or []	×
391	existing_validations = source_config.get("validations") or []	×
392	validations.extend(	×
393	filter(
394	lambda v: v not in existing_validations,
395	runtime_kwargs["validations"],
396	)
397	)
398	dest_config["validations"] = validations	×
399	if runtime_kwargs.get("profilers") is not None:	×
400	profilers = dest_config.get("profilers") or []	×
401	existing_profilers = source_config.get("profilers") or []	×
402	profilers.extend(	×
403	filter(lambda v: v not in existing_profilers, runtime_kwargs["profilers"])
404	)
405	dest_config["profilers"] = profilers	×
406
407	return dest_config	×
408
409
410	def get_updated_action_list(	3✔
411	base_action_list: list, other_action_list: list
412	) -> List[dict]:
413	if base_action_list is None:	×
414	base_action_list = []	×
415
416	base_action_list_dict = {action["name"]: action for action in base_action_list}	×
417
418	for other_action in other_action_list:	×
419	other_action_name = other_action["name"]	×
420	if other_action_name in base_action_list_dict:	×
421	if other_action["action"]:	×
422	nested_update(	×
423	base_action_list_dict[other_action_name],
424	other_action,
425	dedup=True,
426	)
427	else:
428	base_action_list_dict.pop(other_action_name)	×
429	else:
430	base_action_list_dict[other_action_name] = other_action	×
431
432	for other_action in other_action_list:	×
433	other_action_name = other_action["name"]	×
434	if other_action_name in base_action_list_dict:	×
435	if not other_action["action"]:	×
436	base_action_list_dict.pop(other_action_name)	×
437
438	return list(base_action_list_dict.values())	×
439
440
441	def batch_request_in_validations_contains_batch_data(	3✔
442	validations: Optional[List[dict]] = None,
443	) -> bool:
444	if validations is not None:	×
445	for idx, val in enumerate(validations):	×
446	if (	×
447	val.get("batch_request") is not None
448	and isinstance(val.get("batch_request"), (dict, DictDot))
449	and val["batch_request"].get("runtime_parameters") is not None
450	and val["batch_request"]["runtime_parameters"].get("batch_data")
451	is not None
452	):
453	return True	×
454
455	return False	×
456
457
458	def get_validations_with_batch_request_as_dict(	3✔
459	validations: Optional[list] = None,
460	) -> Optional[list]:
461	if validations:	×
462	for value in validations:	×
463	if "batch_request" in value:	×
464	value["batch_request"] = get_batch_request_as_dict(	×
465	batch_request=value["batch_request"]
466	)
467
468	return validations	×
469
470
471	def validate_validation_dict(validation_dict: dict) -> None:	3✔
472	if validation_dict.get("batch_request") is None:	×
473	raise ge_exceptions.CheckpointError("validation batch_request cannot be None")	×
474	if not validation_dict.get("expectation_suite_name"):	×
475	raise ge_exceptions.CheckpointError(	×
476	"validation expectation_suite_name must be specified"
477	)
478	if not validation_dict.get("action_list"):	×
479	raise ge_exceptions.CheckpointError("validation action_list cannot be empty")	×
480
481
482	def send_cloud_notification(url: str, headers: dict):	3✔
483	"""
484	Post a CloudNotificationAction to GE Cloud Backend for processing.
485	"""
486	session = requests.Session()	×
487
488	try:	×
489	response = session.post(url=url, headers=headers)	×
490	except requests.ConnectionError:	×
491	logger.error(f"Failed to connect to Cloud backend after {10} retries.")	×
492	except Exception as e:	×
493	logger.error(str(e))	×
494	else:
495	if response.status_code != 200:	×
496	message = f"Cloud Notification request returned error {response.status_code}: {response.text}"	×
497	logger.error(message)	×
498	return {"cloud_notification_result": message}	×
499	else:
500	return {"cloud_notification_result": "Cloud notification succeeded."}	×
501
502
503	def send_sns_notification(	3✔
504	sns_topic_arn: str, sns_subject: str, validation_results: str, **kwargs
505	) -> str:
506	"""
507	Send JSON results to an SNS topic with a schema of:
508
509
510	:param sns_topic_arn: The SNS Arn to publish messages to
511	:param sns_subject: : The SNS Message Subject - defaults to expectation_suite_identifier.expectation_suite_name
512	:param validation_results: The results of the validation ran
513	:param kwargs: Keyword arguments to pass to the boto3 Session
514	:return: Message ID that was published
515
516	"""
517	if not boto3:	×
518	logger.warning("boto3 is not installed")	×
519	return "boto3 is not installed"	×
520
521	message_dict = {	×
522	"TopicArn": sns_topic_arn,
523	"Subject": sns_subject,
524	"Message": json.dumps(validation_results),
525	"MessageAttributes": {
526	"String": {"DataType": "String.Array", "StringValue": "ValidationResults"},
527	},
528	"MessageStructure": "json",
529	}
530	session = boto3.Session(**kwargs)	×
531	sns = session.client("sns")	×
532	try:	×
533	response = sns.publish(**message_dict)	×
534	except sns.exceptions.InvalidParameterException:	×
535	logger.error(f"Received invalid for message: {validation_results}")	×
536	else:
537	return f"Successfully posted results to {response['MessageId']} with Subject {sns_subject}"	×

great-expectations / great_expectations / 20220908.21

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous