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

How to use _get_column_names method in autotest

Best Python code snippet using autotest_python

db.py

Source:db.py

...7class DataBaseAPI:8    def __init__(self, db_name):9        self.db_name = db_name10        self.table_col_names = {11            "artists": self._get_column_names("artists"),12            "albums": self._get_column_names("albums"),13            "tracks": self._get_column_names("tracks"),14        }15    def _connect(self):16        return sqlite3.connect(self.db_name)17    def _get_column_names(self, table_name):18        query = GET_COL_NAMES_TEMPL.format(table_name)19        with self._connect() as con:20            cur = con.cursor()21            cur.execute(query)22            res = [descr[0] for descr in cur.description if descr[0] != "user_id"]23            cur.close()24        return res25    def _do_get_all(self, table_name, user_id):26        col_names = self.table_col_names[table_name]27        query = GET_ALL_TEMPL.format(", ".join(col_names), table_name)28        with self._connect() as con:29            res = [dict(zip(col_names, row)) for row in con.execute(query, (user_id,)).fetchall()]30        return res31    def _do_add_one(self, table_name, user_id, item):...

base.py

Source:base.py

...40        return {k: getattr(self, k) for k in self.__slots__}41    def to_json(self) -> str:42        return kjson.dumps(self.to_dict())  # type: ignore43    @classmethod44    def _get_column_names(cls) -> List[str]:45        return list(filter(lambda x: x != 'id', cls.__slots__))46    @classmethod47    def get_select_all_query(cls) -> str:48        return f'SELECT * FROM {cls.table_name}'49    @classmethod50    def get_select_one_query(cls) -> str:51        return f'SELECT * FROM {cls.table_name} WHERE id=%(id)s'52    @classmethod53    def get_select_active_query(cls) -> str:54        return f'SELECT * FROM {cls.table_name} WHERE active=TRUE'55    @classmethod56    def get_insert_query(cls) -> str:57        column_names = cls._get_column_names()58        columns = ', '.join(column_names)59        values = ', '.join(f'%({column})s' for column in column_names)60        q = (61            f'INSERT INTO {cls.table_name} ({columns}) '62            f'VALUES ({values}) RETURNING id'63        )64        return q65    @classmethod66    def get_update_query(cls) -> str:67        column_names = cls._get_column_names()68        update_data = ', '.join(f'{column}=%({column})s' for column in column_names)69        return f'UPDATE {cls.table_name} SET {update_data} WHERE id=%(id)s'70    @classmethod71    def get_delete_query(cls) -> str:72        return f'UPDATE {cls.table_name} SET active=FALSE WHERE id=%(id)s'73    def insert(self, curs) -> None:74        curs.execute(self.get_insert_query(), self.to_dict())75        inserted_id, = curs.fetchone()76        setattr(self, 'id', inserted_id)77    def __repr__(self) -> str:...

db_loaders.py

Source:db_loaders.py

...24            self.query = _file.read()25        alias = self.source["database"]26        self.connection = self.db_registry.create_connection_to_database(alias)27        self.cursor = self.connection.cursor()28    def _get_column_names(self):29        return tuple(30            name31            for name, _, _, _, _, _, _32            in self.cursor.description33        )34    def _as_dict(self, column_names, row):35        return dict(zip(column_names, row))36    @helpers.step37    def get_single(self, parameters):38        self.cursor.execute(self.query, parameters)39        column_names = self._get_column_names()40        row = self.cursor.fetchone()41        if row is None:42            raise NoDBRowMatched()43        return self._as_dict(column_names, row)44    @helpers.step45    def get_multiple(self, parameters):46        self.cursor.execute(self.query, parameters)47        column_names = self._get_column_names()48        for row in self.cursor.fetchall():49            yield self._as_dict(column_names, row)50@BaseDbLoader.register_default_noop51class CachedDbLoader(BaseDbLoader):52    def __init__(self, source):53        super(CachedDbLoader, self).__init__()54        self.source = source55        self.cached = self._get_cached_queries(self.source)56    def _get_cached_queries(self, source):57        cached = self.OperationSettings.cached[source['database']]58        for filename, queries in cached.iteritems():59            if source['file'].endswith(filename):60                return queries61    def _get_cached_query(self, parameters):...

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.