KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use exec_sql method in autotest

Best Python code snippet using autotest_python

spatialops.py

Source:spatialops.py

...12    """Executes SQL query and returns DataFrame."""13    conn = loader.database._connection14    return sql.read_frame(query, conn)15    16exec_sql("""17CREATE OR REPLACE FUNCTION expandoverlap_metric(a geometry, b geometry, maxe double precision, maxslice double precision) RETURNS integer AS $BODY$ BEGIN FOR i IN 0..maxslice LOOP IF st_expand(a,maxe*i/maxslice) && b THEN RETURN i; END IF; END LOOP; RETURN 99999999; END; $BODY$ LANGUAGE plpgsql IMMUTABLE COST 100; ALTER FUNCTION expandoverlap_metric(geometry, geometry, double precision, double precision) OWNER TO postgres18""")19if db_to_df("select exists (select 1 from pg_type where typname = 'pgis_nn');").values[0][0] == False:20    exec_sql("CREATE TYPE pgis_nn AS (nn_gid integer, nn_dist numeric(16,5))")21    22exec_sql("""23CREATE OR REPLACE FUNCTION _pgis_fn_nn(geom1 geometry, distguess double precision, numnn integer, maxslices integer, lookupset character varying, swhere character varying, sgid2field character varying, sgeom2field character varying) RETURNS SETOF pgis_nn AS $BODY$ DECLARE strsql text; rec pgis_nn; ncollected integer; it integer; BEGIN ncollected := 0; it := 0; WHILE ncollected < numnn AND it <= maxslices LOOP strsql := 'SELECT currentit.' || sgid2field || ', st_distance(ref.geom, currentit.' || sgeom2field || ') as dist FROM ' || lookupset || '  as currentit, (SELECT geometry(''' || CAST(geom1 As text) || ''') As geom) As ref WHERE ' || swhere || ' AND st_distance(ref.geom, currentit.' || sgeom2field || ') <= ' || CAST(distguess As varchar(200)) || ' AND st_expand(ref.geom, ' || CAST(distguess*it/maxslices As varchar(100)) ||  ') && currentit.' || sgeom2field || ' AND expandoverlap_metric(ref.geom, currentit.' || sgeom2field || ', ' || CAST(distguess As varchar(200)) || ', ' || CAST(maxslices As varchar(200)) || ') = ' || CAST(it As varchar(100)) || ' ORDER BY st_distance(ref.geom, currentit.' || sgeom2field || ') LIMIT ' ||  CAST((numnn - ncollected) As varchar(200)); FOR rec in EXECUTE (strsql) LOOP IF ncollected < numnn THEN ncollected := ncollected + 1; RETURN NEXT rec; ELSE EXIT; END IF; END LOOP; it := it + 1; END LOOP; END $BODY$ LANGUAGE plpgsql STABLE COST 100 ROWS 1000; ALTER FUNCTION _pgis_fn_nn(geometry, double precision, integer, integer, character varying, character varying, character varying, character varying) OWNER TO postgres24""")25exec_sql("""26CREATE OR REPLACE FUNCTION pgis_fn_nn(geom1 geometry, distguess double precision, numnn integer, maxslices integer, lookupset character varying, swhere character varying, sgid2field character varying, sgeom2field character varying) RETURNS SETOF pgis_nn AS $BODY$ SELECT * FROM _pgis_fn_nn($1,$2, $3, $4, $5, $6, $7, $8); $BODY$ LANGUAGE sql STABLE COST 100 ROWS 1000; ALTER FUNCTION pgis_fn_nn(geometry, double precision, integer, integer, character varying, character varying, character varying, character varying) OWNER TO postgres27""")28exec_sql('drop table if exists parcels_outside_taz;')29exec_sql('SELECT * into parcels_outside_taz FROM parcels where taz is null;')30exec_sql("""31update parcels_outside_taz set taz = nn.nn_gid from 32(select parcels_outside_taz.*, 33(pgis_fn_nn(parcels_outside_taz.geom, 10000, 1, 1, 'staging.taz', 'true', 'taz_key', 'geom')).* 34from parcels_outside_taz)35as nn where parcels_outside_taz.gid = nn.gid;36""")...

db_test.py

Source:db_test.py

2from .db import exec_sql3from .db import close_db4connect_db(':memory:')5def _create_test_tables():6    exec_sql('create table test_table (id integer primary key, key text, value text)')7def _drop_test_tables():8    exec_sql('drop table test_table')9def test_no_sql_statment():10    assert exec_sql('') == None11def test_sql_create_and_drop():12    result = exec_sql('create table test_table (id integer primary key, key text, value text)')13    assert result == { 'status': True, 'error': None, 'data': None }14    result = exec_sql('select name from sqlite_master where type="table"')15    assert result == { 'status': True, 'error': None, 'data': [('test_table',)] }16    result = exec_sql('drop table test_table')17    assert result == { 'status': True, 'error': None, 'data': None }18    result = exec_sql('select name from sqlite_master where type="table"')19    assert result == { 'status': True, 'error': None, 'data': [] }20    result = exec_sql('drop table no_test_table')21    assert result == { 'status': False, 'error': 'no such table: no_test_table', 'data': None }22    result = exec_sql('select name from sqlite_master where type="table"')23    assert result == { 'status': True, 'error': None, 'data': [] }24def test_sql_insert_select():25    _create_test_tables()26    result = exec_sql('insert into test_table (key, value) values ("key one", "value one")')27    assert result == { 'status': True, 'error': None, 'data': 1 }28    result = exec_sql('insert into test_table (key, value) values ("key two", "value two")')29    assert result == { 'status': True, 'error': None, 'data': 1 }30    result = exec_sql('select * from test_table')31    assert result == { 'status': True, 'error': None, 'data': [32        (1, 'key one', 'value one'),33        (2, 'key two', 'value two')34    ]}35    _drop_test_tables()36def test_sql_insert_select_binds():37    _create_test_tables()38    data = ['key one', 'value one']39    result = exec_sql('insert into test_table (key, value) values (?, ?)', data)40    assert result == { 'status': True, 'error': None, 'data': 1 }41    data = {'key': 'key two', 'val': 'value two'}42    result = exec_sql('insert into test_table (key, value) values (:key, :val)', data)43    assert result == { 'status': True, 'error': None, 'data': 1 }44    data = [45        ('key three', 'value three'),46        ('key four', 'value four'),47        ('key five', 'value five')48    ]49    result = exec_sql('insert into test_table (key, value) values (?, ?)', data)50    assert result == { 'status': True, 'error': None, 'data': 3 }51    result = exec_sql('select * from test_table')52    assert result == { 'status': True, 'error': None, 'data': [53        (1, 'key one', 'value one'),54        (2, 'key two', 'value two'),55        (3, 'key three', 'value three'),56        (4, 'key four', 'value four'),57        (5, 'key five', 'value five')58    ]}...

db.py

Source:db.py

1import traceback2import sqlite33from utils.log import init_logger4from config.sqlite_config import SqliteConfig5from utils.dependency_injection import provider6from utils.dependency_injection.wiring import provide7provider.assemble(SqliteConfig)8sqlite_config = provide(SqliteConfig)9logger = init_logger(__name__)10class DataBase(object):11    def __init__(self, db_file_path):12        self._db_file_path = db_file_path13    def init_conn(self):14        conn = sqlite3.connect(self._db_file_path)15        return conn16    def init_table_by_mapper(self, mapper):17        conn = self.init_conn()18        try:19            cur = conn.cursor()20            for table_name, fields in mapper.items():21                exec_sql = "CREATE TABLE IF NOT EXISTS %s (" % table_name22                for field in fields:23                    for k, v in field.items():24                        exec_sql = exec_sql + "%s %s," % (k, v)25                exec_sql = exec_sql[:-1] + ")"26                logger.debug(exec_sql)27                cur.execute(exec_sql)28        except BaseException as e:29            traceback.print_exc()30            logger.error(e)31        finally:32            conn.close()33        return self34    def dict_factory(self, cursor, row):35        d = {}36        for index, col in enumerate(cursor.description):37            d[col[0]] = row[index]38        return d39    def select(self, exec_sql):40        conn = self.init_conn()41        conn.row_factory = self.dict_factory42        try:43            cur = conn.cursor()44            logger.debug("exec sql: %s" % exec_sql)45            res = cur.execute(exec_sql).fetchall()46            return res47        except BaseException as e:48            traceback.print_exc()49            logger.error(e)50        finally:51            conn.close()52    def op_with_commit(self, exec_sql):53        conn = self.init_conn()54        cur = conn.cursor()55        logger.debug("exec sql: %s" % exec_sql)56        cur.execute(exec_sql)57        conn.commit()58        logger.debug("affected rows: %s" % cur.rowcount)59        conn.close()60        return cur.rowcount...

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.