Source code for db2rest.db
import sqlalchemy as sql
from sqlalchemy.orm.session import sessionmaker
[docs]class DBAdapter(object):
def __init__(self, db_engine):
self.meta = sql.schema.MetaData()
self.db_engine = db_engine
self.meta.reflect(bind=db_engine)
self.session = sessionmaker(db_engine)()
self.conn = db_engine.connect()
[docs] def add_row(self, table_name, values):
table = sql.Table(table_name, self.meta)
if set(values).issubset(set(table.columns.keys())):
stmt = sql.sql.expression.insert(table, values)
res = self.conn.execute(stmt)
return res.lastrowid
[docs] def delete_row(self, table_name, row_id):
table = sql.Table(table_name, self.meta)
stmt = sql.sql.expression.delete(table).\
where(table.c.id == row_id)
res = self.conn.execute(stmt)
self.session.commit()
return res.rowcount
[docs] def get_tables(self):
return [x.name for x in reversed(self.meta.sorted_tables)]
[docs] def get_rows(self, table_name):
table = sql.Table(table_name, self.meta)
return self.session.query(table).all()
[docs] def get_row(self, table_name, row_id):
"""Returns a list with a row found.
"""
table = sql.Table(table_name, self.meta)
return [self.session.query(table).filter_by(id=row_id).one()]
[docs] def update_row(self, table_name, row_id, values):
"""Update the given row_id in the given table.
"""
table = sql.Table(table_name, self.meta)
stmt = sql.sql.expression.update(table).\
where(table.c.id == row_id).\
values(values)
res = self.conn.execute(stmt)
self.session.commit()
return res.rowcount,\
dict(zip(self.get_headers(table_name),
self.get_row(table_name, row_id)[0]))