Today I tried Bonobo + SQLAlchemy (for the first time), so it's bound to be full of bad practices. However, this is working code. I'm hoping for some feedback to for doing it correctly.
I started out using the SQLAlchemy extension for Bonobo, but I ran into limitations, so I wrote my own loader. My DB is SQLite3, so it doesn't support multiple connections.
What this does is read from one table, do a simple transformation and load into a target table with a different table structure.
SQLite database:
CREATE TABLE sys_batch (
batch_id INTEGER PRIMARY KEY AUTOINCREMENT,
batch_name VARCHAR2 ( 100 ) );
CREATE TABLE sys_batch_tgt (
batch_id INTEGER PRIMARY KEY,
name_of_batch VARCHAR2 ( 100 ) );
Python code:
import sqlalchemy
from sqlalchemy.ext.automap import automap_base
from sqlalchemy.sql import select
from sqlalchemy.orm import Session
import bonobo
#import bonobo_sqlalchemy
engine = sqlalchemy.create_engine('sqlite:///C:/Py/bonobo/coolstuff.db')
Base = automap_base()
Base.prepare(engine, reflect=True)
# Tables
sys_batch = Base.classes.sys_batch
sys_batch_tgt = Base.classes.sys_batch_tgt
#def get_services():
# return {'sqlalchemy.pgengine': sqlalchemy.create_engine('sqlite:///C:/Py/bonobo/coolstuff.db')}
def extract_sys_batch():
conn = engine.connect()
s = select([sys_batch])
result = conn.execute(s)
#conn.close()
row = result.fetchone()
yield row
# convert incoming SQLAlchemy "ProxyRow" to a dict, so we can manipulate it.
def tfm_0(row):
tgt_row = {}
tgt_row['batch_id'] = row['batch_id']
tgt_row['name_of_batch'] = row['batch_name']
return tgt_row
def tfm_1(row):
row['batch_id'] = row['batch_id'] + 100
return row
def ldr_1(row):
# write to target
session = Session(engine)
thing = sys_batch_tgt(**row)
session.add(thing)
session.commit()
def get_graph(**options):
return bonobo.Graph(
extract_sys_batch,
tfm_0,
tfm_1,
ldr_1,
bonobo.PrettyPrinter(),
)
# The __main__ block actually execute the graph.
if __name__ == '__main__':
parser = bonobo.get_argument_parser()
#each transformation is its own thread, so we have to open/close in there.
#conn = engine.connect()
with bonobo.parse_args(parser) as options:
bonobo.run(get_graph(**options)) #, services=get_services(**options)
#conn.close()
get_graph
ignores all options? \$\endgroup\$ – Reinderien Dec 19 '18 at 14:59