SQLAlchemyを使ってOracleのデータをSQLiteにコピー (試作)

OracleのデータをSQLiteにコピーしていじりかったところにSQLAlchemyがあったので金槌の法則で使ってみました。

SQLAlchemyの0.9.8で試しています。

import re

from sqlalchemy import create_engine
from sqlalchemy.ext.automap import automap_base
from sqlalchemy.dialects.sqlite.base import SQLiteDDLCompiler, SQLiteTypeCompiler

class Oracle2SQLiteTypeCompiler(SQLiteTypeCompiler):
    '''とりいそぎ必要だったカラムだけ考える'''
    def visit_DOUBLE_PRECISION(self, type_):
        pass
    def visit_INTERVAL(self, type_):
        pass
    def visit_LONG(self, type_):
        pass
    def visit_NUMBER(self, type_, **kw):
        return self.visit_NUMERIC(type_)
    def visit_NVARCHAR2(self, type_):
        return self.visit_NVARCHAR(type_)
    def visit_RAW(self, type_):
        pass
    def visit_ROWID(self, type_):
        pass
    def visit_VARCHAR2(self, type_):
        return self.visit_VARCHAR(type_)

TODATE_REGEXP = re.compile("to_date\('([^']+)'")

class Oracle2SQLiteDDLCompiler(SQLiteDDLCompiler):
    '''とりいそぎ必要だったカラムだけ考える'''
    def get_column_default_string(self, column):
        default_clause = super().get_column_default_string(column)
        if default_clause is None:
            return default_clause
        default_clause_lower = default_clause.lower()
        if 'sysdate' in default_clause_lower:
            return 'current_date'
        m = TODATE_REGEXP.match(default_clause_lower)
        if m:
            return "'%s'" % m.group(1)
        return default_clause

from_engine = create_engine('oracle://FROM_DB_USER:PASSWORD@HSOT:PORT/SID')
to_engine = create_engine('sqlite:///to_db.db')

to_engine.dialect.type_compiler = Oracle2SQLiteTypeCompiler(to_engine.dialect)
to_engine.dialect.ddl_compiler = Oracle2SQLiteDDLCompiler

base = automap_base()
base.prepare(from_engine, reflect=True)

table = base.metadata.tables['db_table_name']
table.create(to_engine)

data = from_engine.execute(table.select()).fetchall()
to_engine.execute(table.insert(), data)