How to use check_values method in Slash

Best Python code snippet using slash

sql.py

Source:sql.py Github

copy

Full Screen

1#2# Licensed to the Apache Software Foundation (ASF) under one3# or more contributor license agreements. See the NOTICE file4# distributed with this work for additional information5# regarding copyright ownership. The ASF licenses this file6# to you under the Apache License, Version 2.0 (the7# "License"); you may not use this file except in compliance8# with the License. You may obtain a copy of the License at9#10# http://www.apache.org/licenses/LICENSE-2.011#12# Unless required by applicable law or agreed to in writing,13# software distributed under the License is distributed on an14# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY15# KIND, either express or implied. See the License for the16# specific language governing permissions and limitations17# under the License.18from typing import Any, Dict, Optional, Union19from airflow.exceptions import AirflowException20from airflow.operators.sql import BaseSQLOperator21def parse_boolean(val: str) -> Union[str, bool]:22 """Try to parse a string into boolean.23 Raises ValueError if the input is not a valid true- or false-like string value.24 """25 val = val.lower()26 if val in ('y', 'yes', 't', 'true', 'on', '1'):27 return True28 if val in ('n', 'no', 'f', 'false', 'off', '0'):29 return False30 raise ValueError(f"{val!r} is not a boolean-like string value")31def _get_failed_tests(checks):32 return [33 f"\tCheck: {check},\n\tCheck Values: {check_values}\n"34 for check, check_values in checks.items()35 if not check_values["success"]36 ]37class SQLColumnCheckOperator(BaseSQLOperator):38 """39 Performs one or more of the templated checks in the column_checks dictionary.40 Checks are performed on a per-column basis specified by the column_mapping.41 Each check can take one or more of the following options:42 - equal_to: an exact value to equal, cannot be used with other comparison options43 - greater_than: value that result should be strictly greater than44 - less_than: value that results should be strictly less than45 - geq_to: value that results should be greater than or equal to46 - leq_to: value that results should be less than or equal to47 - tolerance: the percentage that the result may be off from the expected value48 :param table: the table to run checks on49 :param column_mapping: the dictionary of columns and their associated checks, e.g.50 .. code-block:: python51 {52 "col_name": {53 "null_check": {54 "equal_to": 0,55 },56 "min": {57 "greater_than": 5,58 "leq_to": 10,59 "tolerance": 0.2,60 },61 "max": {"less_than": 1000, "geq_to": 10, "tolerance": 0.01},62 }63 }64 :param conn_id: the connection ID used to connect to the database65 :param database: name of database which overwrite the defined one in connection66 .. seealso::67 For more information on how to use this operator, take a look at the guide:68 :ref:`howto/operator:SQLColumnCheckOperator`69 """70 column_checks = {71 "null_check": "SUM(CASE WHEN column IS NULL THEN 1 ELSE 0 END) AS column_null_check",72 "distinct_check": "COUNT(DISTINCT(column)) AS column_distinct_check",73 "unique_check": "COUNT(column) - COUNT(DISTINCT(column)) AS column_unique_check",74 "min": "MIN(column) AS column_min",75 "max": "MAX(column) AS column_max",76 }77 def __init__(78 self,79 *,80 table: str,81 column_mapping: Dict[str, Dict[str, Any]],82 conn_id: Optional[str] = None,83 database: Optional[str] = None,84 **kwargs,85 ):86 super().__init__(conn_id=conn_id, database=database, **kwargs)87 for checks in column_mapping.values():88 for check, check_values in checks.items():89 self._column_mapping_validation(check, check_values)90 self.table = table91 self.column_mapping = column_mapping92 # OpenLineage needs a valid SQL query with the input/output table(s) to parse93 self.sql = f"SELECT * FROM {self.table};"94 def execute(self, context=None):95 hook = self.get_db_hook()96 failed_tests = []97 for column in self.column_mapping:98 checks = [*self.column_mapping[column]]99 checks_sql = ",".join([self.column_checks[check].replace("column", column) for check in checks])100 self.sql = f"SELECT {checks_sql} FROM {self.table};"101 records = hook.get_first(self.sql)102 if not records:103 raise AirflowException(f"The following query returned zero rows: {self.sql}")104 self.log.info(f"Record: {records}")105 for idx, result in enumerate(records):106 tolerance = self.column_mapping[column][checks[idx]].get("tolerance")107 self.column_mapping[column][checks[idx]]["result"] = result108 self.column_mapping[column][checks[idx]]["success"] = self._get_match(109 self.column_mapping[column][checks[idx]], result, tolerance110 )111 failed_tests.extend(_get_failed_tests(self.column_mapping[column]))112 if failed_tests:113 raise AirflowException(114 f"Test failed.\nQuery:\n{self.sql}\nResults:\n{records!s}\n"115 "The following tests have failed:"116 f"\n{''.join(failed_tests)}"117 )118 self.log.info("All tests have passed")119 def _get_match(self, check_values, record, tolerance=None) -> bool:120 match_boolean = True121 if "geq_to" in check_values:122 if tolerance is not None:123 match_boolean = record >= check_values["geq_to"] * (1 - tolerance)124 else:125 match_boolean = record >= check_values["geq_to"]126 elif "greater_than" in check_values:127 if tolerance is not None:128 match_boolean = record > check_values["greater_than"] * (1 - tolerance)129 else:130 match_boolean = record > check_values["greater_than"]131 if "leq_to" in check_values:132 if tolerance is not None:133 match_boolean = record <= check_values["leq_to"] * (1 + tolerance) and match_boolean134 else:135 match_boolean = record <= check_values["leq_to"] and match_boolean136 elif "less_than" in check_values:137 if tolerance is not None:138 match_boolean = record < check_values["less_than"] * (1 + tolerance) and match_boolean139 else:140 match_boolean = record < check_values["less_than"] and match_boolean141 if "equal_to" in check_values:142 if tolerance is not None:143 match_boolean = (144 check_values["equal_to"] * (1 - tolerance)145 <= record146 <= check_values["equal_to"] * (1 + tolerance)147 ) and match_boolean148 else:149 match_boolean = record == check_values["equal_to"] and match_boolean150 return match_boolean151 def _column_mapping_validation(self, check, check_values):152 if check not in self.column_checks:153 raise AirflowException(f"Invalid column check: {check}.")154 if (155 "greater_than" not in check_values156 and "geq_to" not in check_values157 and "less_than" not in check_values158 and "leq_to" not in check_values159 and "equal_to" not in check_values160 ):161 raise ValueError(162 "Please provide one or more of: less_than, leq_to, "163 "greater_than, geq_to, or equal_to in the check's dict."164 )165 if "greater_than" in check_values and "less_than" in check_values:166 if check_values["greater_than"] >= check_values["less_than"]:167 raise ValueError(168 "greater_than should be strictly less than "169 "less_than. Use geq_to or leq_to for "170 "overlapping equality."171 )172 if "greater_than" in check_values and "leq_to" in check_values:173 if check_values["greater_than"] >= check_values["leq_to"]:174 raise ValueError(175 "greater_than must be strictly less than leq_to. "176 "Use geq_to with leq_to for overlapping equality."177 )178 if "geq_to" in check_values and "less_than" in check_values:179 if check_values["geq_to"] >= check_values["less_than"]:180 raise ValueError(181 "geq_to should be strictly less than less_than. "182 "Use leq_to with geq_to for overlapping equality."183 )184 if "geq_to" in check_values and "leq_to" in check_values:185 if check_values["geq_to"] > check_values["leq_to"]:186 raise ValueError("geq_to should be less than or equal to leq_to.")187 if "greater_than" in check_values and "geq_to" in check_values:188 raise ValueError("Only supply one of greater_than or geq_to.")189 if "less_than" in check_values and "leq_to" in check_values:190 raise ValueError("Only supply one of less_than or leq_to.")191 if (192 "greater_than" in check_values193 or "geq_to" in check_values194 or "less_than" in check_values195 or "leq_to" in check_values196 ) and "equal_to" in check_values:197 raise ValueError(198 "equal_to cannot be passed with a greater or less than "199 "function. To specify 'greater than or equal to' or "200 "'less than or equal to', use geq_to or leq_to."201 )202class SQLTableCheckOperator(BaseSQLOperator):203 """204 Performs one or more of the checks provided in the checks dictionary.205 Checks should be written to return a boolean result.206 :param table: the table to run checks on207 :param checks: the dictionary of checks, e.g.:208 .. code-block:: python209 {210 "row_count_check": {"check_statement": "COUNT(*) = 1000"},211 "column_sum_check": {"check_statement": "col_a + col_b < col_c"},212 }213 :param conn_id: the connection ID used to connect to the database214 :param database: name of database which overwrite the defined one in connection215 .. seealso::216 For more information on how to use this operator, take a look at the guide:217 :ref:`howto/operator:SQLTableCheckOperator`218 """219 sql_check_template = "CASE WHEN check_statement THEN 1 ELSE 0 END AS check_name"220 sql_min_template = "MIN(check_name)"221 def __init__(222 self,223 *,224 table: str,225 checks: Dict[str, Dict[str, Any]],226 conn_id: Optional[str] = None,227 database: Optional[str] = None,228 **kwargs,229 ):230 super().__init__(conn_id=conn_id, database=database, **kwargs)231 self.table = table232 self.checks = checks233 # OpenLineage needs a valid SQL query with the input/output table(s) to parse234 self.sql = f"SELECT * FROM {self.table};"235 def execute(self, context=None):236 hook = self.get_db_hook()237 check_names = [*self.checks]238 check_mins_sql = ",".join(239 self.sql_min_template.replace("check_name", check_name) for check_name in check_names240 )241 checks_sql = ",".join(242 [243 self.sql_check_template.replace("check_statement", value["check_statement"]).replace(244 "check_name", check_name245 )246 for check_name, value in self.checks.items()247 ]248 )249 self.sql = f"SELECT {check_mins_sql} FROM (SELECT {checks_sql} FROM {self.table});"250 records = hook.get_first(self.sql)251 if not records:252 raise AirflowException(f"The following query returned zero rows: {self.sql}")253 self.log.info(f"Record: {records}")254 for check in self.checks.keys():255 for result in records:256 self.checks[check]["success"] = parse_boolean(str(result))257 failed_tests = _get_failed_tests(self.checks)258 if failed_tests:259 raise AirflowException(260 f"Test failed.\nQuery:\n{self.sql}\nResults:\n{records!s}\n"261 "The following tests have failed:"262 f"\n{', '.join(failed_tests)}"263 )...

Full Screen

Full Screen

98-validate-binary-search-tree.py

Source:98-validate-binary-search-tree.py Github

copy

Full Screen

1# first solution2class Solution:3 def check_values(self, root, value, direction):4 print(root.val, value, direction)5 if direction == "right" and root.val <= value:6 return False7 elif direction == "left" and root.val >= value:8 return False9 if root.left and root.right:10 return (11 self.check_values(root.left, value, direction) and 12 self.check_values(root.right, value, direction)13 )14 elif root.left:15 return self.check_values(root.left, value, direction)16 elif root.right:17 return self.check_values(root.right, value, direction)18 else:19 return True20 21 def isValidBST(self, root: Optional[TreeNode]) -> bool:22 if root.left and root.right:23 return (24 self.check_values(root.left, root.val, "left") and 25 self.check_values(root.right, root.val, "right") and26 self.isValidBST(root.left) if root.val > root.left.val else False and27 self.isValidBST(root.right) if root.val < root.right.val else False28 )29 elif root.left:30 return (31 self.check_values(root.left, root.val, "left") and32 self.isValidBST(root.left) if root.val > root.left.val else False33 )34 elif root.right:35 return (36 self.check_values(root.right, root.val, "right") and37 self.isValidBST(root.right) if root.val < root.right.val else False38 )39 else:...

Full Screen

Full Screen

math.py

Source:math.py Github

copy

Full Screen

1import civilpy2np = civilpy.np3solver = civilpy.sym4inf = civilpy.sym.oo5def equation_solver(eq, variable):6 return civilpy.sym.solve(eq, variable)7def arithmetic_growth(p_0, r, t):8 p_t = p_0 + (t * r)9 return p_t10def geometric_growth(p_0, r, t):11 p_t = p_0 * ((1 + r) ** t)12 return p_t13def find_global_extrema_order_3(f, x1, x2, variable='x'):14 x = civilpy.sym.symbols(variable)15 f_prime = civilpy.sym.diff(f)16 f_prime2 = civilpy.sym.diff(f_prime)17 print(f'The first derivative is \n{f_prime} the second is \n {f_prime2}\n')18 try:19 root1, root2 = civilpy.sym.solvers.solve(f_prime, x)20 print(f'The first root is {root1}, the second root is {root2}\n')21 for root in [root1, root2]:22 if f_prime2.subs(x, root) > 0:23 print(f'{root} is a local minimum\n')24 elif f_prime2.subs(x, root) < 0:25 print(f'{root} is a local maximum\n')26 inf_point = civilpy.sym.solvers.solve(f_prime2, x)27 print(f'The inflection point is {inf_point[0]}\n')28 check_values = {x1: 0, x2: 0, root1: 0, root2: 0}29 for key, value in check_values.items():30 check_values[key] = f.subs(x, key)31 global_min = min(check_values.keys(), key=(lambda k: check_values[k]))32 global_max = max(check_values.keys(), key=(lambda k: check_values[k]))33 print(f'Global min: {global_min}, Global max: {global_max}')34 return global_min, global_max, check_values35 except ValueError:...

Full Screen

Full Screen

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run Slash automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful