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

How to use as_sql method in autotest

Best Python code snippet using autotest_python

sql.py

Source:sql.py

...120        self.where = where121        self.order_by = order_by122        self.limit = limit123        124    def as_sql(self):125        query_table = self.generate_query_field(self.table)126        query_fields = self.generate_query_fields(self.fields)127        query_where = self.generate_query_where(self.where)128        query_order_by = self.generate_query_order_by(self.order_by)129        query_limit = self.generate_query_limit(self.limit)130    131        extra_query = ' '.join([132            i for i in [query_where, query_order_by, query_limit] if i133        ])134        self.query = self.BASE_QUERY.format(135            query_fields, query_table, extra_query136        )137        return self.query, self.params138    139    140class DeleteStatement(BaseSQLStatement):141    BASE_QUERY = 'DELETE FROM {} {}'142    def __init__(self, table, where=None):143        super().__init__()144        self.table = table145        self.where = where146    147    def as_sql(self):148        query_table = self.generate_query_field(self.table)149        query_where = self.generate_query_where(self.where)150        extra_query = query_where151    152        self.query = self.BASE_QUERY.format(153            query_table, extra_query154        )155        return self.query, self.params156    157    158class CreateStatement(BaseSQLStatement):159    BASE_QUERY = '{} INTO {} ({}) VALUES ({})'160    161    def __init__(self, table, assignments, mode, compress_fields):162        super().__init__()163        self.table = table164        self.assignments = assignments165        self.mode = mode166        self.compress_fields = compress_fields167        168    def generate_query_mode(self, mode):169        if mode == 'INSERT':170            return 'INSERT'171        elif mode == 'IGNORE':172            return 'INSERT IGNORE'173        elif mode == 'REPLACE':174            return 'REPLACE'175        176    def generate_query_fields_values(self, assignments, compress_fields):177        fields = [a.field for a in assignments]178        query_values = ['%s']*len(assignments)179        if type(compress_fields) == list:180            for index, field in enumerate(fields):181                for cf in compress_fields:182                    if '`'+cf+'`' == field:  # field should be like '`id`'183                        query_values[index] = 'COMPRESS(%s)'184        query_fields = ', '.join(fields)185        query_values = ', '.join(query_values)186        for a in assignments:187            self.params.append(a.value)188        return query_fields, query_values189    190    def as_sql(self):191        192        query_mode = self.generate_query_mode(self.mode)193        194        query_table = self.generate_query_field(self.table)195        196        query_fields, query_values = self.generate_query_fields_values(197            self.assignments, self.compress_fields198        )199        200        self.query = self.BASE_QUERY.format(201            query_mode, query_table, query_fields, query_values202        )203        return self.query, self.params204    205    206class UpdateStatement(BaseSQLStatement):207    BASE_QUERY = 'UPDATE {} SET {} {}'208    209    def __init__(self, table, assignments, where=None):210        super().__init__()211        self.table = table212        self.assignments = assignments213        self.where = where214        215    def generate_query_fields_values(self, assignments):216        query = ', '.join(217            [a.field + '=%s' for a in assignments]218        )219        for a in assignments:220            self.params.append(a.value)221        return query222    def as_sql(self):223        query_table = self.generate_query_field(self.table)224    225        query_fields_values = self.generate_query_fields_values(226            self.assignments)227        query_where = self.generate_query_where(self.where)228    229        self.query = self.BASE_QUERY.format(230            query_table, query_fields_values, query_where231        )232        return self.query, self.params233    234def where_clauses_from_filters(filters):235    where_clauses = []236    for op, k, v in filters:237        where_clause = WhereClause(k, op, v)238        where_clauses.append(where_clause)239    return where_clauses240def assignment_clauses_clauses_from_filters(data):241    assignment_clauses = []242    for k, v in data.items():243        clause = AssignmentClause(k, v)244        assignment_clauses.append(clause)245    return assignment_clauses246def query_parameters_from_create(collection, data, mode='INSERT', compress_fields=None):247    table = FieldClause(collection)248    assignments = assignment_clauses_clauses_from_filters(data)249    query, params = CreateStatement(table, assignments, mode, compress_fields).as_sql()250    return query, params251def query_parameters_from_update(collection, filters, data):252    table = FieldClause(collection)253    assignments = assignment_clauses_clauses_from_filters(data)254    where = where_clauses_from_filters(filters)255    query, params = UpdateStatement(table, assignments, where).as_sql()256    return query, params257def query_parameters_from_get(collection, filters, fields=None):258    table = FieldClause(collection)259    where = where_clauses_from_filters(filters)260    fields = [FieldClause(f) for f in (fields or [])]261    query, params = SelectStatement(table, fields, where, limit=1).as_sql()262    return query, params263def query_parameters_from_delete(collection, filters):264    table = FieldClause(collection)265    where = where_clauses_from_filters(filters)266    query, params = DeleteStatement(table, where).as_sql()267    return query, params268def query_parameters_from_filter(269        collection, filters, fields=None, order_by=None, limit=None):270    table = FieldClause(collection)271    where = where_clauses_from_filters(filters)272    fields = [FieldClause(f) for f in (fields or [])]273    274    if order_by is None:275        order_by = []276    elif isinstance(order_by, str):277        order_by = [order_by]278    order_by = [FieldClause(f) for f in order_by]279    280    query, params = SelectStatement(281        table, fields, where, order_by, limit).as_sql()...

compiler.py

Source:compiler.py

...25            return [(OrderBy(Value("__scatter__"), descending=descending), False)]26        return super(SQLCompiler, self).find_ordering_name(27            name, opts, alias=alias, default_order=default_order, already_seen=already_seen28        )29    def as_sql(self, with_limits=True, with_col_aliases=False, subquery=False):30        self.pre_sql_setup()31        self.refcounts_before = self.query.alias_refcount.copy()32        select = SelectCommand(self.connection, self.query)33        return (select, tuple())34    def get_select(self):35        self.query.select_related = False  # Make sure select_related is disabled for all queries36        return super(SQLCompiler, self).get_select()37class SQLInsertCompiler(SQLCompiler, compiler.SQLInsertCompiler):38    def __init__(self, *args, **kwargs):39        self.return_id = None40        super(SQLInsertCompiler, self).__init__(*args, **kwargs)41    def as_sql(self, with_limits=True, with_col_aliases=False, subquery=False):42        self.pre_sql_setup()43        from gcloudc.db.backends.datastore.utils import get_concrete_fields44        # Always pass down all the fields on an insert45        return [46            (47                InsertCommand(48                    self.connection,49                    self.query.model,50                    self.query.objs,51                    list(self.query.fields) + list(get_concrete_fields(self.query.model, ignore_leaf=True)),52                    self.query.raw,53                ),54                tuple(),55            )56        ]57class SQLDeleteCompiler(SQLCompiler, compiler.SQLDeleteCompiler):58    def as_sql(self, with_limits=True, with_col_aliases=False, subquery=False):59        return (DeleteCommand(self.connection, self.query), tuple())60class SQLUpdateCompiler(SQLCompiler, compiler.SQLUpdateCompiler):61    def as_sql(self, with_limits=True, with_col_aliases=False, subquery=False):62        self.pre_sql_setup()63        return (UpdateCommand(self.connection, self.query), tuple())64class SQLAggregateCompiler(compiler.SQLAggregateCompiler, SQLCompiler):65    def as_sql(self, with_limits=True, with_col_aliases=False, subquery=False):66        if self.query.subquery:67            self.query.high_mark = self.query.subquery.query.high_mark68            self.query.low_mark = self.query.subquery.query.low_mark...

where.pyi

Source:where.pyi

...14    default: Any = ...15    resolved: bool = ...16    conditional: bool = ...17    def split_having(self, negated: bool = ...) -> Tuple[Optional[WhereNode], Optional[WhereNode]]: ...18    def as_sql(self, compiler: SQLCompiler, connection: Any) -> Any: ...19    def get_group_by_cols(self) -> List[Expression]: ...20    def relabel_aliases(self, change_map: Union[Dict[Optional[str], str], OrderedDict]) -> None: ...21    def clone(self) -> WhereNode: ...22    def relabeled_clone(self, change_map: Union[Dict[Optional[str], str], OrderedDict]) -> WhereNode: ...23    def resolve_expression(self, *args: Any, **kwargs: Any) -> WhereNode: ...24class NothingNode:25    contains_aggregate: bool = ...26    def as_sql(self, compiler: SQLCompiler = ..., connection: Any = ...) -> Any: ...27class ExtraWhere:28    contains_aggregate: bool = ...29    sqls: List[str] = ...30    params: Optional[Union[List[int], List[str]]] = ...31    def __init__(self, sqls: List[str], params: Optional[Union[List[int], List[str]]]) -> None: ...32    def as_sql(self, compiler: SQLCompiler = ..., connection: Any = ...) -> Tuple[str, Union[List[int], List[str]]]: ...33class SubqueryConstraint:34    contains_aggregate: bool = ...35    alias: str = ...36    columns: List[str] = ...37    targets: List[str] = ...38    query_object: Query = ...39    def __init__(self, alias: str, columns: List[str], targets: List[str], query_object: Query) -> None: ......

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.