Skip to content

Instantly share code, notes, and snippets.

@Razzo78
Last active January 26, 2018 09:29
Show Gist options
  • Save Razzo78/3580cf9ed1f3c9a34484ab887b6dbf5d to your computer and use it in GitHub Desktop.
Save Razzo78/3580cf9ed1f3c9a34484ab887b6dbf5d to your computer and use it in GitHub Desktop.
Python, Sqlite - Import database and mapping tables to class and execute a query to populate mapped class
from sqlalchemy import create_engine, MetaData, Table
from sqlalchemy.orm import mapper, sessionmaker
from sqlalchemy.sql import select
class Table1Name(object):
pass
class Table2Name(object):
pass
class DbImport:
def __init__(self, name):
self.engine = create_engine('sqlite:///%s' % name, echo=False)
metadata = MetaData(self.engine)
self.table_1_name = Table('Table1Name', metadata, autoload=True)
mapper(Table1Name, self.table_1_name)
self.table_2_name = Table('Table2Name', metadata, autoload=True)
mapper(Table1Name, self.table_2_name)
def load_session(self):
session = sessionmaker(bind=self.engine)
return session()
def get_id_table_1_name_by_name(self, name):
conn = self.engine.connect()
s = select([Table1Name.id]).where(Table1Name.name == name)
res = conn.execute(s)
row = res.fetchone()
return None if row is None else row['id']
def exist_table_2_name(self, id_table_2_name):
conn = self.engine.connect()
s = select([Table2Name.id]).where(Table2Name.id == id_table_2_name)
res = conn.execute(s)
row = res.fetchone()
return False if row is None else True
def insert_new_table_1_name(self, dict_data):
ins = self.table_1_name.insert(). \
values(dict_data)
# print(ins.compile().params)
conn = self.engine.connect()
return conn.execute(ins)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment