TestMu Conference is Back! Future of Testing + AIRegister for Free

How to use print_queries method in autotest

Best Python code snippet using autotest_python

list.py

Source:list.py

1"""2Functions to support list script.3"""4from sqlalchemy import create_engine5from sqlalchemy.orm import sessionmaker6from sqlalchemy import func7import sqlparse8from modelmeta import DataFile, DataFileVariable, Ensemble, TimeSet9# argument parser helpers10def strtobool(string):11    return string.lower() in {'true', 't', 'yes', '1'}12main_arg_names = '''13    print_queries14    count15    ensemble16    multi_variable17    multi_year_mean18    mym_concatenated19'''.split()20def print_query(title, query):21    print('--', title)22    compiled_query = str(23        query.statement.compile(compile_kwargs={'literal_binds': True})24    )25    formatted_query = sqlparse.format(26        compiled_query, reindent=True, keyword_case='upper')27    print(formatted_query)28    print()29def list_information(query, template, count=False):30    if count:31        print(query.count())32    else:33        results = query.all()34        for row in results:35            print(template.format(row=row))36def data_file_query(37        session,38        print_queries=False,39        ensemble=None,40        multi_variable=None,41        multi_year_mean=None,42        mym_concatenated=None,43):44    """45    Select DataFiles matching selection criteria46    :param session:47    :param print_queries:48    :param ensemble:49    :param multi_variable:50    :param multi_year_mean:51    :param mym_concatenated:52    :return:53    """54    query = session.query(DataFile.id)55    if ensemble:56        query = (57            query.join(DataFile.data_file_variables)58                .join(DataFileVariable.ensembles)59                .filter(Ensemble.name == ensemble)60        )61    if multi_variable is not None:62        query = (63            query.join(DataFile.data_file_variables)64                .group_by(DataFile.id)65        )66        if multi_variable:67            query = query.having(func.count(DataFileVariable.id) > 1)68        else:69            query = query.having(func.count(DataFileVariable.id) == 1)70    if multi_year_mean is not None:71        query = (72            query.join(TimeSet)73                .filter(TimeSet.multi_year_mean == multi_year_mean)74        )75    if mym_concatenated is not None:76        query = (77            query.join(TimeSet)78                .filter(TimeSet.multi_year_mean == True)79        )80        if mym_concatenated:81            query = query.filter(TimeSet.num_times.in_((5, 13, 16, 17)))82        else:83            query = query.filter(TimeSet.num_times.in_((1, 4, 12)))84    if print_queries:85        print_query('Data file query', query)86    return query87def _list_filepaths(88        session,89        print_queries=False,90        count=False,91        ensemble=None,92        multi_variable=None,93        multi_year_mean=None,94        mym_concatenated=None,95        list_ensembles = None,96):97    df_query = data_file_query(98        session,99        print_queries=print_queries,100        ensemble=ensemble,101        multi_variable=multi_variable,102        multi_year_mean=multi_year_mean,103        mym_concatenated=mym_concatenated,104    )105    if list_ensembles:106        if count:107            info_query = (108                session.query(109                    Ensemble.name.label('ensemble_name'),110                    func.count(DataFile.id).label('number')111                )112                    .filter(DataFile.id.in_(df_query))113                    .join(DataFile.data_file_variables)114                    .join(DataFileVariable.ensembles)115                    .group_by(Ensemble.name)116            )117        else:118            info_query = (119                session.query(120                    DataFile.filename,121                    func.string_agg(Ensemble.name, ',').label('ensemble_names')122                )123                    .filter(DataFile.id.in_(df_query))124                    .join(DataFile.data_file_variables)125                    .join(DataFileVariable.ensembles)126                    .group_by(DataFile.id)127            )128    else:129        info_query = df_query.add_columns(DataFile.filename)130    if print_queries:131        print_query('Info query', info_query)132    if list_ensembles:133        template = '{row.filename}\t{row.ensemble_names}'134    else:135        template = '{row.filename}'136    list_information(info_query, template, count=count)137def _list_dirpaths(138        session,139        print_queries=False,140        count=False,141        ensemble=None,142        multi_variable=None,143        multi_year_mean=None,144        mym_concatenated=None,145        depth=999,146):147    df_query = data_file_query(148        session,149        print_queries=print_queries,150        ensemble=ensemble,151        multi_variable=multi_variable,152        multi_year_mean=multi_year_mean,153        mym_concatenated=mym_concatenated,154    )155    info_query = (156        session.query(157            func.regexp_replace(158                DataFile.filename,159                r'^((/.[^/]+){{1,{}}}/).+$'.format(depth),160                r'\1'161            ).label('dir_path'),162            func.count().label('number')163        )164            .filter(DataFile.id.in_(df_query))165            .group_by('dir_path')166            .order_by('dir_path')167    )168    if print_queries:169        print_query('Info query', info_query)170    template = '{row.dir_path} ({row.number})'171    list_information(info_query, template, count=count)172def list_filepaths(args):173    arg_names = main_arg_names + '''174        list_ensembles175    '''.split()176    engine = create_engine(args.dsn)177    session = sessionmaker(bind=engine)()178    _list_filepaths(session, **{key: getattr(args, key) for key in arg_names})179def list_dirpaths(args):180    arg_names = main_arg_names + '''181        depth182    '''.split()183    engine = create_engine(args.dsn)184    session = sessionmaker(bind=engine)()...

aws.py

Source:aws.py

1"""2Production configuration for grading controller3"""4import json5import os6from .logsettings import get_logger_config7from .settings import *8import logging9######################################################################10#General config11######################################################################12#Allow to specify a prefix for env/auth configuration files13SERVICE_VARIANT = os.environ.get('SERVICE_VARIANT', '')14if SERVICE_VARIANT:15    CONFIG_PREFIX = SERVICE_VARIANT + "."16#Read config from json file17with open(ENV_ROOT / CONFIG_PREFIX + "env.json") as env_file:18    ENV_TOKENS = json.load(env_file)19#Debug20DEBUG = ENV_TOKENS.get('DEBUG', False)21if isinstance(DEBUG,basestring):22    DEBUG= DEBUG.lower()=="true"23TEMPLATE_DEBUG = ENV_TOKENS.get('TEMPLATE_DEBUG', False)24if isinstance(TEMPLATE_DEBUG,basestring):25    TEMPLATE_DEBUG= TEMPLATE_DEBUG.lower()=="true"26PRINT_QUERIES = ENV_TOKENS.get('PRINT_QUERIES', False)27if isinstance(PRINT_QUERIES,basestring):28    PRINT_QUERIES= PRINT_QUERIES.lower()=="true"29#General30REQUESTS_TIMEOUT = int(ENV_TOKENS.get('REQUESTS_TIMEOUT', REQUESTS_TIMEOUT))31TIME_BETWEEN_XQUEUE_PULLS = int(TIME_BETWEEN_XQUEUE_PULLS)32TIME_BETWEEN_EXPIRED_CHECKS = int(ENV_TOKENS.get('TIME_BETWEEN_EXPIRED_CHECKS', TIME_BETWEEN_EXPIRED_CHECKS))33GRADER_SETTINGS_DIRECTORY = ENV_TOKENS.get('GRADER_SETTINGS_DIRECTORY')34MAX_NUMBER_OF_TIMES_TO_RETRY_GRADING = int(ENV_TOKENS.get('MAX_NUMBER_OF_TIMES_TO_RETRY_GRADING'))35DEFAULT_ESTIMATED_GRADING_TIME = int(ENV_TOKENS.get('DEFAULT_ESTIMATED_GRADING_TIME', DEFAULT_ESTIMATED_GRADING_TIME))36COURSE_DATA_PATH = ENV_TOKENS.get('COURSE_DATA_PATH', COURSE_DATA_PATH)37#ML38MIN_TO_USE_ML = int(ENV_TOKENS.get('MIN_TO_USE_ML', MIN_TO_USE_ML))39ML_MODEL_PATH = os.path.join(ENV_ROOT, ENV_TOKENS.get('ML_MODEL_PATH'))40TIME_BETWEEN_ML_CREATOR_CHECKS = int(ENV_TOKENS.get('TIME_BETWEEN_ML_CREATOR_CHECKS', TIME_BETWEEN_ML_CREATOR_CHECKS))41TIME_BETWEEN_ML_GRADER_CHECKS = int(TIME_BETWEEN_ML_GRADER_CHECKS)42USE_S3_TO_STORE_MODELS= ENV_TOKENS.get('USE_S3_TO_STORE_MODELS', USE_S3_TO_STORE_MODELS)43if isinstance(USE_S3_TO_STORE_MODELS,basestring):44    USE_S3_TO_STORE_MODELS= USE_S3_TO_STORE_MODELS.lower()=="true"45S3_BUCKETNAME=ENV_TOKENS.get('S3_BUCKETNAME',"OpenEnded")46#Peer47MIN_TO_USE_PEER = int(ENV_TOKENS.get('MIN_TO_USE_PEER', MIN_TO_USE_PEER))48PEER_GRADER_COUNT = int(ENV_TOKENS.get('PEER_GRADER_COUNT', PEER_GRADER_COUNT))49PEER_GRADER_MINIMUM_TO_CALIBRATE = int(ENV_TOKENS.get("PEER_GRADER_MINIMUM_TO_CALIBRATE", PEER_GRADER_MINIMUM_TO_CALIBRATE))50PEER_GRADER_MAXIMUM_TO_CALIBRATE = int(ENV_TOKENS.get("PEER_GRADER_MAXIMUM_TO_CALIBRATE", PEER_GRADER_MAXIMUM_TO_CALIBRATE))51PEER_GRADER_MIN_NORMALIZED_CALIBRATION_ERROR = float(ENV_TOKENS.get("PEER_GRADER_MIN_NORMALIZED_CALIBRATION_ERROR", PEER_GRADER_MIN_NORMALIZED_CALIBRATION_ERROR))52PEER_GRADE_FINISHED_SUBMISSIONS_WHEN_NONE_PENDING = ENV_TOKENS.get("PEER_GRADE_FINISHED_SUBMISSIONS_WHEN_NONE_PENDING",53                                                                   PEER_GRADE_FINISHED_SUBMISSIONS_WHEN_NONE_PENDING)54#Submission Expiration55EXPIRE_SUBMISSIONS_AFTER = int(ENV_TOKENS.get('EXPIRE_SUBMISSIONS_AFTER', EXPIRE_SUBMISSIONS_AFTER))56RESET_SUBMISSIONS_AFTER = int(ENV_TOKENS.get('RESET_SUBMISSIONS_AFTER', RESET_SUBMISSIONS_AFTER))57GENERATE_COURSE_DATA_EVERY = int(ENV_TOKENS.get('GENERATE_COURSE_DATA_EVERY', 12 * 60 * 60))58#Time zone (shows up in logs)59TIME_ZONE = ENV_TOKENS.get('TIME_ZONE', TIME_ZONE)60local_loglevel = ENV_TOKENS.get('LOCAL_LOGLEVEL', 'INFO')61LOG_DIR = ENV_TOKENS.get("LOG_DIR", ENV_ROOT / "log")62LOGGING = get_logger_config(debug=DEBUG)63######################################################################64# Read secure config65with open(ENV_ROOT / CONFIG_PREFIX + "auth.json") as auth_file:66    AUTH_TOKENS = json.load(auth_file)67XQUEUE_INTERFACE = AUTH_TOKENS['XQUEUE_INTERFACE']68GRADING_CONTROLLER_INTERFACE = AUTH_TOKENS['GRADING_CONTROLLER_INTERFACE']69DATABASES = AUTH_TOKENS['DATABASES']70AWS_ACCESS_KEY_ID = AUTH_TOKENS.get("AWS_ACCESS_KEY_ID","")...

learn_db.py

Source:learn_db.py

...17            price_case=Case(low_price_case, high_price_case, output_field=TextField())18        ).annotate(19            discount_price=Case(low_price, high_price, output_field=DecimalField())20        ).order_by('category_id', 'name').select_related('category')21        print(print_queries('SELECT', connection.queries))22        t_list = PrettyTable(["ÐÐ°ÑÐµÐ³Ð¾ÑÐ¸Ñ", "Ð¢Ð¾Ð²Ð°Ñ", "ÐÐºÑÐ¸Ñ", 'Ð¦ÐµÐ½Ð°', 'Ð¦ÐµÐ½Ð° ÑÐ¾ ÑÐºÐ¸Ð´ÐºÐ¾Ð¹'])23        t_list.align = 'l'24        i = 025        for product in filter_products:26            t_list.add_row([f'{product.category.name:15}', f'{product.name:15}', f'{product.price_case:5}',27                            f'{product.price:6.0f}',28                            f'{product.discount_price:6.0f}'])29            if i == 0:30                print(print_queries('SELECT', connection.queries))31                i += 1...

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.