Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use templated_query method in pyresttest

Best Python code snippet using pyresttest_python

bigquery_validator.py

Source:bigquery_validator.py

...45        local_config_path = os.path.join(os.getcwd(), 'bq_validator_config.json')46        local_config_params = load_config_from_file(local_config_path)47        params = {**params, **local_config_params}48        return params49    def render_templated_query(self, templated_query):50        """Convert the Jinja templated SQL to a valid query"""51        templated_query = templated_query.replace('params.', '')  # need this to get formatting correct52        t = Template(templated_query)53        rendered_query = t.render(self.params)54        rendered_query = rendered_query.replace('\n', ' ')55        return rendered_query56    def render_templated_query_from_file(self, file_path):57        """Convert the Jinja templated SQL to a valid query"""58        templated_query = read_sql_file(file_path)59        return self.render_templated_query(templated_query)60    # # todo finish61    # def parameterize_sql(self, query):62    #     default_params = get_default_params()63    #     for k, v in self.params.items():64    #         # default params do not require 'param.' prefix65    #         if k in default_params:66    #             query = query.replace(f'{{{{  {k} }}}}', v)67    #         else:68    #             query = query.replace(f'{{{{ params.{k} }}}}', v)69    #     return query70    def dry_run_query(self, query):71        """Run a BigQuery query with dry_run set to True.72        If the query succeeds it is valid and will return the estimated processing bytes required for the query.73        An exception will be thrown if the query is not valid.74        """75        try:76            job_config = bigquery.QueryJobConfig(dry_run=True, use_query_cache=self.use_query_cache)77            # Start the query, passing in the extra configuration.78            query_job = self.bq_client.query(79                query,80                job_config=job_config,81            )  # Make an API request.82            # A dry run query completes immediately.83            total_bytes = query_job.total_bytes_processed84            logging.info("This query will process {} bytes.".format(total_bytes))85            byte = 186            kilobyte = 102487            megabyte = 1024 * 102488            gigabyte = 1024 * 1024 * 102489            terabyte = 1024 * 1024 * 1024 * 102490            query_cost = {91                'b': round(total_bytes / byte, 2),92                'kb': round(total_bytes / kilobyte, 2),93                'mb': round(total_bytes / megabyte, 2),94                'gb': round(total_bytes / gigabyte, 2),95                'tb': round(total_bytes / terabyte, 2)96            }97            if total_bytes > terabyte:98                rounded_total = query_cost['b']99                byte_type = 'TB'100            elif total_bytes > gigabyte:101                rounded_total = query_cost['gb']102                byte_type = 'GB'103            elif total_bytes > megabyte:104                rounded_total = query_cost['mb']105                byte_type = 'MB'106            elif total_bytes > kilobyte:107                rounded_total = query_cost['kb']108                byte_type = 'KB'109            else:110                rounded_total = query_cost['b']111                byte_type = 'B'112            if self.return_query_cost_as_dict:113                return True, query_cost114            else:115                message = f'This query will process {rounded_total} {byte_type}.'116                return True, message117        except Exception as e:118            error_string = str(e)119            error_minus_job_url = error_string.split('jobs?prettyPrint=false:')120            split_error = error_minus_job_url[1].split('\n\n')121            syntax_error = split_error[0]122            return False, syntax_error123    # TODO: validate output of query124    # Stolen from Apache Beam125    # https://github.com/apache/beam/blob/87e11644c44a4c677ec2faa78f50cdffbb33605a/sdks/python/apache_beam/io/gcp/tests/bigquery_matcher.py126    # @retry.with_exponential_backoff(127    #       num_retries=MAX_RETRIES,128    #       retry_filter=retry_on_http_timeout_and_value_error)129    #   def run_query(self):130    #     """Run Bigquery query with retry if got error http response"""131    #     _LOGGER.info('Attempting to perform query %s to BQ', self.query)132    #     # Create client here since it throws an exception if pickled.133    #     bigquery_client = bigquery.Client(self.project)134    #     query_job = bigquery_client.query(self.query)135    #     rows = query_job.result(timeout=60)136    #     return [row.values() for row in rows]137    def validate_query(self, templated_query):138        """Check if query passed as parameter is valid. If the query contains any Jinja templated params they will139        be converted to the associated param value if one exists.140        Parameters:141        templated_query (str): SQL query to be validated142        """143        try:144            formatted_query = self.render_templated_query(templated_query)145            querv_is_valid, message = self.dry_run_query(formatted_query)146            logging.info(f'Query is { "valid" if querv_is_valid else "invalid"}. {message}')147            return querv_is_valid, message148        except Exception as e:149            logging.error(e)150            return False, f"An error occurred while validating query - {templated_query}"151    def validate_query_from_file(self,152                                 file_path:str,153                                 ignore_leading_lines:int=None):154        """Same as validate_query() but reads query from a file rather than accepting it as a param155        Parameters:156        file_path (str): Path to the sql file on the file system157        ignore_leading_lines (int): Ignore leading n lines from query when validating query158        """159        try:160            # todo check if file ends with .sql161            if os.path.isfile(file_path):162                templated_query = read_sql_file(file_path, ignore_leading_lines)163                return self.validate_query(templated_query)164            else:165                raise ValueError(f'Error: File does not exist: {file_path}')166        except Exception as e:167            logging.error(e)168            return False169    def auto_validate_query_from_file(self,170                                      file_path:str,171                                      ignore_leading_lines:int=None):172        """Continuously monitor a sql file and automatically validate the sql on every saved change to the file.173        Any Jinja templated params will be automatically parsed on update.174        Parameters:175        file_path (str): Path to the sql file on the file system176        ignore_leading_lines (int): Ignore leading n lines from query when validating query177       """178        try:179            _cached_stamp = 0180            while True:181                # TODO monitor query validator config for changes too182                stamp = os.stat(file_path).st_mtime183                if stamp != _cached_stamp:184                    print(f'Loading...{RESET_SEQ}', end='                                                          \r')185                    _cached_stamp = stamp186                    templated_query = read_sql_file(file_path, ignore_leading_lines)187                    formatted_query = self.render_templated_query(templated_query)188                    querv_is_valid, message = self.dry_run_query(formatted_query)189                    logging.info(f'Query is {"valid" if querv_is_valid else "invalid"}')190                    if querv_is_valid:191                        # Extra white space here is quick workaround to remove all text from last message192                        # when the previous message length exceeds the new message length193                        print_success(f'Valid query. {message}{RESET_SEQ}',194                                      end='                                                                         \r')195                    else:196                        print_failure(f'Invalid query. {message}{RESET_SEQ}',197                                      end='                                                                         \r')198        except Exception as e:199            logging.error(e)...

test_sql.py

Source:test_sql.py

1from ..fixtures import mock_whale_dir, mock_template_dir2from pathlib import Path3from mock import call, patch4import pytest5from whale.utils.sql import (6    template_query,7    _validate_and_print_result,8)9VALID_TEMPLATE = "{% set main = 5 %}"10VALID_FILE_NAME = "valid.sql"11INVALID_TEMPLATE = "{% notvalidtext %}"12INVALID_FILE_NAME = "invalid.sql"13QUERY_WITHOUT_TEMPLATE = "hello"14def write_template_to_file(path: Path, content: str):15    with open(path, "w") as f:16        f.write(content)17class TestTemplateDirectory:18    @pytest.fixture(autouse=True)19    def setup(self, mock_template_dir):20        mock_template_dir.mkdir(exist_ok=True)21        write_template_to_file(mock_template_dir / VALID_FILE_NAME, VALID_TEMPLATE)22        write_template_to_file(mock_template_dir / INVALID_FILE_NAME, INVALID_TEMPLATE)23    @patch("builtins.print")24    def test__validate_and_print_result_returns_correct_passing_result(25        self, mock_print, mock_template_dir26    ):27        error = _validate_and_print_result(mock_template_dir / VALID_FILE_NAME)28        mock_print.assert_called_with("[\x1b[32mâ\x1b[0m] valid.sql")29        assert error is None30    @patch("builtins.print")31    def test__validate_and_print_result_returns_correct_failing_result(32        self, mock_print, mock_template_dir33    ):34        error = _validate_and_print_result(mock_template_dir / INVALID_FILE_NAME)35        mock_print.assert_has_calls([call("[\x1b[31mx\x1b[0m] invalid.sql")]),36        assert error is not None37    def test_template_query_is_no_op_without_jinja(self):38        templated_query = template_query(QUERY_WITHOUT_TEMPLATE)39        assert templated_query == QUERY_WITHOUT_TEMPLATE40    def test_template_query_templates_empty_correctly(self):41        templated_query = template_query(VALID_TEMPLATE)42        assert templated_query == ""43    def test_template_query_templates_correctly_with_file(self):44        query = "{{ main }}"45        templated_query = template_query(46            query,47            connection_name="valid",48        )...

patents_view.py

Source:patents_view.py

1from airflow.models import BaseOperator2from airflow.utils.decorators import apply_defaults3from hooks.patents_view import PatentsViewHook4from airflow.utils.helpers import parse_template_string5import json6class PatentsToLocalOperator(BaseOperator):7    """Queries PatentsView API and dumps to local json file.8    Attributes:9        entity: string, name of PatentsView endpoint to query10        query_file_path: string, full local file path to JSON file containing the query parameters11        response_file_path: string, full local file path to write response12    """13    template_fields = ['response_file_path']14    @apply_defaults15    def __init__(self,16                 entity,17                 query_file_path,18                 response_file_path,19                 *args, **kwargs):20        super().__init__(*args, **kwargs)21        self.entity = entity22        self.query_file_path = query_file_path23        self.response_file_path = response_file_path24    def execute(self, context):25        print(f'Querying PatentsView API for {self.entity} with parameters in {self.query_file_path}')26        27        # read in query json file28        with open(self.query_file_path, 'r') as f:29            query = json.load(f)30        # manually render airflow macro in json file31        _, template = parse_template_string(json.dumps(query))32        templated_query = template.render(**context)33        # init hook and post to api34        hook = PatentsViewHook()35        response = hook.post(self.entity, json.loads(templated_query))36        37        with open(self.response_file_path, 'w') as f:38            json.dump(response, f)...

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.