Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use _get_engine method in toolium

Best Python code snippet using toolium_python

example_sqlite.py

Source:example_sqlite.py

...5log = logging.getLogger(__name__)6class DatastoreExampleSqliteBackend(DatastoreBackend):7    def __init__(self):8        self._engine = None9    def _get_engine(self):10        if not self._engine:11            self._engine = create_engine(self.write_url)12        return self._engine13    def _insert_records(self, table, records):14        if len(records):15            for record in records:16                self._get_engine().execute(17                    u'INSERT INTO "{0}"({1}) VALUES({2})'.format(18                        table,19                        u', '.join(record.keys()),20                        u', '.join(['?'] * len(record.keys()))21                    ),22                    record.values()23                )24            pass25    def configure(self, config):26        self.write_url = config.get(27            u'ckan.datastore.write_url'28        ).replace(u'example-', u'')29        return config30    def create(self, context, data_dict):31        columns = str(u', '.join(32            map(lambda e: e['id'] + u' text', data_dict['fields'])))33        engine = self._get_engine()34        engine.execute(35            u' CREATE TABLE IF NOT EXISTS "{name}"({columns});'.format(36                name=data_dict['resource_id'],37                columns=columns38            ))39        self._insert_records(data_dict['resource_id'], data_dict['records'])40        return data_dict41    def upsert(self, context, data_dict):42        raise NotImplementedError()43    def delete(self, context, data_dict):44        engine = self._get_engine()45        engine.execute(u'DROP TABLE IF EXISTS "{0}"'.format(46            data_dict['resource_id']47        ))48        return data_dict49    def search(self, context, data_dict):50        engine = self._get_engine()51        result = engine.execute(u'SELECT * FROM "{0}" LIMIT {1}'.format(52            data_dict['resource_id'],53            data_dict.get(u'limit', 10)54        ))55        data_dict['records'] = map(dict, result.fetchall())56        data_dict['total'] = len(data_dict['records'])57        fields_info = []58        for name, type in self.resource_fields(59                data_dict['resource_id'])['schema'].items():60            fields_info.append({61                u'type': type,62                u'id': name63            })64        data_dict['fields'] = fields_info65        return data_dict66    def search_sql(self, context, data_dict):67        raise NotImplementedError()68    def make_private(self, context, data_dict):69        pass70    def make_public(self, context, data_dict):71        pass72    def resource_exists(self, id):73        return self._get_engine().execute(74            u'''75            select name from sqlite_master76            where type = "table" and name = "{0}"'''.format(77                id)78        ).fetchone()79    def resource_fields(self, id):80        engine = self._get_engine()81        info = engine.execute(82            u'PRAGMA table_info("{0}")'.format(id)).fetchall()83        schema = {}84        for col in info:85            schema[col.name] = col.type86        return {u'schema': schema, u'meta': {}}87    def resource_id_from_alias(self, alias):88        if self.resource_exists(alias):89            return True, alias90        return False, alias91    def get_all_ids(self):92        return map(lambda t: t.name, self._get_engine().execute(93            u'''94            select name from sqlite_master95            where type = "table"'''...

sqla_utils.py

Source:sqla_utils.py

1from sqlalchemy import create_engine2from core.libs.config_controller import get_config3def _get_engine(echo=False):4    """ Retrieves the database engine (SQLAlchemy - sync).5    :return: the database engine6    """7    db_config = get_config().database8    engine = getattr(_get_engine, '_db_engine', None)9    if engine is None:10        engine = create_engine(db_config.dsn)11        setattr(_get_engine, '_db_engine', engine)12    engine.echo = echo13    return engine14class SQLAUtils(object):15    """ Class that provides generic methods to work with SQLAlchemy. """...

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.