Skip to content

Instantly share code, notes, and snippets.

@toyg
Last active August 29, 2015 14:27
Show Gist options
  • Save toyg/9fb541ff3dbc8c175329 to your computer and use it in GitHub Desktop.
Save toyg/9fb541ff3dbc8c175329 to your computer and use it in GitHub Desktop.
from sqlalchemy import create_engine, MetaData, Table, PrimaryKeyConstraint
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.dialects.oracle import RAW, NUMBER
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.types import BINARY
@compiles(RAW, "oracle")
def compile_raw_oracle(type_, compiler, **kw):
if type_.length:
return "RAW(%(length)s)" % {'length': type_.length}
else:
return "RAW(2000)"
@compiles(NUMBER, "oracle")
def compile_number_oracle(type_, compiler, **kw):
if 'precision' not in kw:
return "NUMBER(*,0)"
return compiler._generate_numeric(type_, "NUMBER", **kw)
def generic_mapper(table):
Base = declarative_base()
class GenericMapper(Base):
__table__ = table
return GenericMapper
def nopk_mapper(table):
Base = declarative_base()
class GenericMapper(Base):
__table__ = table
__mapper_args__ = {
'primary_key': [list(table.columns)[0]]
}
return GenericMapper
def make_session(connection_string):
engine = create_engine(connection_string, echo=False, convert_unicode=True, auto_setinputsizes=False)
Session = sessionmaker(bind=engine)
return Session(), engine
def copy(from_db, to_db, tables, prefix='TMP_'):
source, sengine = make_session(from_db)
smeta = MetaData(bind=sengine)
destination, dengine = make_session(to_db)
destination.autoflush = False
dmeta = MetaData(bind=dengine)
for table_name in tables:
print('Processing ' + table_name)
print('Reading source table definition')
table = Table(table_name, smeta, autoload=True)
target_name = prefix + str(table.name)
target_table = table.tometadata(dmeta, name=target_name)
for constraint in target_table.constraints:
constraint.name = None
print('Creating destination table')
target_table.metadata.create_all(dengine)
NewRecord = None
if not table.primary_key.name:
NewRecord = nopk_mapper(table)
else:
NewRecord = generic_mapper(target_table)
columns = table.columns.keys()
print('Transferring records')
for record in source.query(table).all():
data = dict(
[(str(column), getattr(record, column)) for column in columns]
)
destination.merge(NewRecord(**data))
print('Committing changes')
destination.commit()
source.close()
destination.close()
if __name__ == '__main__':
from_db='oracle+cx_oracle://scott:tiger@localhost:1521/XE'
to_db='oracle+cx_oracle://scott:tiger@localhost:1521/XE'
copy(from_db, to_db, ['MY_TABLE'])
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment