Best Python code snippet using pytest-benchmark
migration.py
Source:migration.py
...23 new_version)24 _apply_update(instance.engine, new_version)25 session.add(SchemaChanges(schema_version=new_version))26 _LOGGER.info("Upgrade to version %s done", new_version)27def _create_index(engine, table_name, index_name):28 """Create an index for the specified table.29 The index name should match the name given for the index30 within the table definition described in the models31 """32 from sqlalchemy import Table33 from . import models34 table = Table(table_name, models.Base.metadata)35 _LOGGER.debug("Looking up index for table %s", table_name)36 # Look up the index object by name from the table is the the models37 index = next(idx for idx in table.indexes if idx.name == index_name)38 _LOGGER.debug("Creating %s index", index_name)39 index.create(engine)40 _LOGGER.debug("Finished creating %s", index_name)41def _apply_update(engine, new_version):42 """Perform operations to bring schema up to date."""43 if new_version == 1:44 _create_index(engine, "events", "ix_events_time_fired")45 elif new_version == 2:46 # Create compound start/end index for recorder_runs47 _create_index(engine, "recorder_runs", "ix_recorder_runs_start_end")48 # Create indexes for states49 _create_index(engine, "states", "ix_states_last_updated")50 _create_index(engine, "states", "ix_states_entity_id_created")51 else:52 raise ValueError("No schema migration defined for version {}"53 .format(new_version))54def _inspect_schema_version(engine, session):55 """Determine the schema version by inspecting the db structure.56 When the schema verison is not present in the db, either db was just57 created with the correct schema, or this is a db created before schema58 versions were tracked. For now, we'll test if the changes for schema59 version 1 are present to make the determination. Eventually this logic60 can be removed and we can assume a new db is being created.61 """62 from sqlalchemy.engine import reflection63 from .models import SchemaChanges, SCHEMA_VERSION64 inspector = reflection.Inspector.from_engine(engine)...
custom_mongo_client.py
Source:custom_mongo_client.py
...23 """24 Ð¡Ð¾Ð·Ð´Ð°ÐµÑ Ð½ÐµÐ¾Ð±Ñ
одимÑе индекÑÑ Ð² базе даннÑÑ
.25 :param str db_name: Ð¸Ð¼Ñ Ð±Ð°Ð·Ñ Ð´Ð°Ð½Ð½ÑÑ
, в коÑоÑой необÑ
одимо ÑоздаÑÑ Ð¸Ð½Ð´ÐµÐºÑÑ26 """27 self._create_index(db_name, 'imports', IndexModel([('import_id', 1)], unique=True))28 self._create_index(db_name, 'imports', IndexModel([('citizens.citizen_id', 1)]))29 self._create_index(db_name, 'imports', IndexModel([('import_id', 1), ('citizens.citizen_id', 1)], unique=True))30 self._create_index(db_name, 'birthdays', IndexModel([('import_id', 1)], unique=True))31 self._create_index(db_name, 'percentile_age', IndexModel([('import_id', 1)], unique=True))32 def _create_index(self, db_name: str, collection_name: str, index: IndexModel):33 """34 Ð¡Ð¾Ð·Ð´Ð°ÐµÑ Ð¸Ð½Ð´ÐµÐºÑ Ð² Ñказанной коллекÑии Ñказанной Ð±Ð°Ð·Ñ Ð´Ð°Ð½Ð½ÑÑ
.35 ÐÑи налиÑии индекÑа Ñ Ñаким же именем, но дÑÑгими паÑамеÑÑами, ÑдалÑÐµÑ Ð¸Ð¼ÐµÑÑийÑÑ Ð¸Ð½Ð´ÐµÐºÑ Ð¸ ÑÐ¾Ð·Ð´Ð°ÐµÑ Ð½Ð¾Ð²Ñй.36 :param str db_name: Ð¸Ð¼Ñ Ð±Ð°Ð·Ñ Ð´Ð°Ð½Ð½ÑÑ
37 :param str collection_name: Ð¸Ð¼Ñ ÐºÐ¾Ð»Ð»ÐµÐºÑии38 :param IndexModel index: ÑоздаваемÑй индекÑ39 """40 try:41 self[db_name][collection_name].create_indexes([index])42 except OperationFailure:43 self[db_name][collection_name].drop_index(index.document['name'])...
201506081244_a2dfbb4b85c_add_extra_indexes_for_user_search.py
Source:201506081244_a2dfbb4b85c_add_extra_indexes_for_user_search.py
...8from indico.core.db.sqlalchemy.util.queries import has_extension9# revision identifiers, used by Alembic.10revision = 'a2dfbb4b85c'11down_revision = '4074211727ba'12def _create_index(has_trgm, table, column):13 col_func = 'indico_unaccent(lower({}))'.format(column)14 kwargs = {}15 if has_trgm:16 kwargs = {'postgresql_using': 'gin',17 'postgresql_ops': {col_func: 'gin_trgm_ops'}}18 op.create_index(op.f('ix_{}_{}_unaccent'.format(table, column)), table, [sa.text(col_func)], schema='users',19 **kwargs)20def upgrade():21 if context.is_offline_mode():22 raise Exception('This upgrade is only possible in online mode')23 has_trgm = has_extension(op.get_bind(), 'pg_trgm')24 if has_trgm:25 print 'pg_trgm extension is available - creating trigram indexes'26 else:27 print 'pg_trgm extension is not available - creating normal indexes'28 _create_index(has_trgm, 'users', 'first_name')29 _create_index(has_trgm, 'users', 'last_name')30 _create_index(has_trgm, 'users', 'phone')31 _create_index(has_trgm, 'users', 'address')32 _create_index(has_trgm, 'affiliations', 'name')33 _create_index(has_trgm, 'emails', 'email')34def downgrade():35 op.drop_index('ix_users_first_name_unaccent', 'users', schema='users')36 op.drop_index('ix_users_last_name_unaccent', 'users', schema='users')37 op.drop_index('ix_users_phone_unaccent', 'users', schema='users')38 op.drop_index('ix_users_address_unaccent', 'users', schema='users')39 op.drop_index('ix_affiliations_name_unaccent', 'affiliations', schema='users')...
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!!