Last active
January 26, 2018 09:29
-
-
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
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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