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: ......
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!