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

How to use get_table_schema method in localstack

Best Python code snippet using localstack_python

sqldb.py

Source:sqldb.py

...113    return tuple(_col)114def _drop_create_table(tname):115    cursor = m_conn.cursor()116    cursor.execute('DROP TABLE IF EXISTS ' + tname)117    cursor.execute('CREATE TABLE {} ({})'.format(tname, str(get_table_schema(tname))))118    m_logger.info('initialized table: ' + tname)119def create(table, lenient=False, **kwargs) -> TableSchema:120    schema = get_table_schema(table)121    if not lenient:122        try:123            keys = table_keys_dict(table, kwargs, schema)124            assert not existing(table, **keys), f"{keys} already exists at {table}"125        except KeyError:126            pass127    record = schema.new(**kwargs)128    sql = 'INSERT INTO {} ({}) VALUES ({})'.format(table, *record.for_insert())129    m_logger.debug(sql)130    cursor = m_conn.cursor()131    cursor.execute(sql)132    m_conn.commit()133    m_logger.info(f'created at {table} {repr(record)}')134    return record135def update(table, **kwargs):136    schema = get_table_schema(table)137    keys = table_keys_dict(table, kwargs, schema)138    if not existing(table, **keys):139        raise NameError(f"table {table} is missing {keys}")140    record = schema.new(**kwargs)141    where = schema.new(**keys).for_where(**keys)142    _set = record.for_update(**kwargs)143    sql = f'UPDATE {table} SET {_set} WHERE {where}'144    cursor = m_conn.cursor()145    cursor.execute(sql)146    m_conn.commit()147    m_logger.debug(f'updated at {table} {sql}')148def read(table, **kv) -> TableSchema:149    where = get_table_schema(table).new(**kv).for_where(**kv)150    sql = f"SELECT * FROM {table} WHERE {where}"151    m_logger.debug('reading: ' + sql)152    try:153        values = next(_select(sql))154    except StopIteration:155        raise NameError('missing from {}: {}={} "{}"'.format(table, *list(kv.items())[0], sql))156    record = _new_schema(table, values)157    m_logger.debug(f'read from {table} {repr(record)}')158    return record159def existing(table, by_schema=True, **where) -> bool:160    if by_schema:161        by_schema = get_table_schema(table)162        where_sql = by_schema.new(**where).for_where(**where)163    else:164        where_sql = ' AND '.join(f'{k}{where_op_value(str(v))}'165                                 for k, v in where.items() if v)166    sql = f"SELECT 1 FROM {table} WHERE {where_sql} LIMIT 1"167    try:168        values = next(_select(sql))169    except StopIteration:170        values = None171    except sqlite3.OperationalError as exc:172        values = None173        if str(exc) != f'no such table: {table}':174            raise exc175    exists = values is not None and len(values) > 0176    m_logger.debug(table + ' ' +177                   ' '.join(f'{k}={v}' for k, v in where.items()) +178                   f" {'does' if exists else 'does not'} exist")179    return exists180def write(table, **kwargs):181    try:182        update(table, **kwargs)183    except NameError:184        create(table, lenient=True, **kwargs)185    except KeyError as exc:186        if exc.args[0] != '__key__':187            raise188        delete(table, lenient=True, **kwargs)189        create(table, **kwargs)190def delete(table, lenient=False, by_schema=True, **where):191    if by_schema:192        by_schema = get_table_schema(table)193        for_where = by_schema.new(**where).for_where(**where)194    else:195        for_where = ' '.join(f"{k}={_quoted(v)}" for k, v in where.items())196    sql = f'DELETE FROM {table}'197    if not lenient and where:198        assert existing(table, **where), f"table {table} is missing {for_where}"199    if where:200        sql += ' WHERE ' + for_where201    cursor = m_conn.cursor()202    cursor.execute(sql)203    m_conn.commit()204    m_logger.debug('Done ' + sql)205def list_table(table, **where) -> Iterator:206    return (_new_schema(table, row) for row in rows(table, **where))207def select(table: str, *columns, by_schema=True, **where) -> Iterable:  # yield row208    sql = f"SELECT {','.join(columns) if columns else '*'} FROM {table}"209    if 'order_by' in where:210        order_by = ' ORDER BY ' + where['order_by']211        del where['order_by']212    else:213        order_by = ''214    if where:215        sql += ' WHERE '216        if by_schema:217            sql += get_table_schema(table).new(**where).for_where(**where)218        else:219            sql += ' '.join(f"{k}={_quoted(v)}" for k, v in where.items())220    sql += order_by221    for row in _select(sql):222        yield row223def _select(sql) -> Iterable:  # yield row224    m_logger.debug(sql)225    m_conn.commit()226    cursor = m_conn.cursor()227    try:228        cursor.execute(sql)229    except Exception as exc:230        raise type(exc)(str(exc) + f' "{sql}"')231    row = cursor.fetchone()232    while row:233        yield row234        row = cursor.fetchone()235def select_join(left: str, right: str, on: str) -> Iterable:  # yield row236    sql = 'SELECT * FROM ' + left + ' LEFT JOIN ' + right + ' ON ' + '{}.{} = {}.{}'.format(left, on, right, on)237    for row in _select(sql):238        yield row239def select_objects(table: str, *columns, **where) -> Iterable:  # (OrderedAttrDict, )240    return (OrderedAttrDict(zip(list(f for f in get_table_schema(table).keys() if not columns or f in columns), row))241            for row in select(table, *columns, **where))242def select_join_objects(left: str, right: str, on: str) -> Iterable:  # (OrderedAttrDict, )243    return (244        OrderedAttrDict(zip(list(get_table_schema(left).keys())[:-1] + list(get_table_schema(right).keys())[:-1], row))245        for row in select_join(left, right, on)246    )247def rows(table: str, sep: str = '', **where) -> Iterable:248    return (sep.join(row) if sep else row249            for row in select(table, **where))250def _new_schema(table, values) -> TableSchema:251    return get_table_schema(table).new(**dict(zip(m_table_columns[table].names, values)))252def _new_dump_file(out: str, cwd: str = '', table: str = '') -> Dumper:253    fmt = dump_file_fmt(out)254    return DUMPERS[fmt](name=dump_file_path(out, cwd, table))255def dump_file_path(out: str, cwd: str = '', table: str = '') -> str:256    fmt = dump_file_fmt(out)257    return os.path.join(cwd, out.replace(fmt, table + '.' + fmt))258def _close_dump_files(files: [Dumper]) -> [str]:259    closed = []260    for file in files:261        file.close()262        closed.append(file.name)263    return closed264def dump(*outs, cwd: str = '') -> [str]:265    assert outs, 'expected one or more out files, formats: ' + ', '.join(DUMPERS.keys())266    dumped = []267    tables = list(get_table_schemas().keys())268    for table in tables:269        dump_files = [_new_dump_file(out, cwd, table) for out in outs]270        for row in select_objects(table):271            [file.dump(get_table_schema(table).new(**row)) for file in dump_files]272        dumped.extend(_close_dump_files(dump_files))...

config_table.py

Source:config_table.py

1import sqlite3 2def get_table_schema(conn, table_name):3    schema = []4    ins = "PRAGMA table_info(\'" + table_name + "\')"5    print(ins)6    for row in conn.execute(ins).fetchall():7        schema.append(row[1])8    return schema9def rename_columns(request):10    response = dict()11    conn = sqlite3.connect("gcbm.db") 12    for table, config in request.items():13        response[table] = dict()14        print("TABLE IS ")15        print(table)16        print(config)17        response[table]['schema_before'] = get_table_schema(conn, table)18        for old_name, new_name in config.items():19            print(old_name, new_name)20            try:21                rename = 'ALTER TABLE '  + table +  ' RENAME COLUMN ' +  old_name + ' TO ' + new_name22                conn.execute(rename)23            except Exception:24                print("Exception occured is ", Exception)25        response[table]['schema_after'] = get_table_schema(conn, table)26        print(response)27    conn.commit()28    conn.close()...

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.