db_copy.py 1.2 KB

1234567891011121314151617181920212223242526272829303132
  1. import sqlalchemy.dialects.mssql.base
  2. from sqlalchemy import create_engine, MetaData, Boolean, DATETIME
  3. from database import conn_string
  4. source_dsn = {'user': 'sa', 'pass': 'Mffu3011#', 'server': 'GC-SERVER1\\GLOBALCUBE', 'database': 'OPTIMA', 'driver': 'mssql'}
  5. source_schema = 'import'
  6. target_dsn = {'user': 'root', 'pass': '', 'server': 'localhost', 'database': 'OPTIMA', 'driver': 'mysql'}
  7. target_schema = ''
  8. source_db = create_engine(conn_string(source_dsn))
  9. source_meta = MetaData()
  10. source_meta.reflect(bind=source_db, schema=source_schema)
  11. # print('\n'.join(source_meta.tables.keys()))
  12. target_db = create_engine(conn_string(target_dsn))
  13. source_meta.bind = target_db
  14. for table in source_meta.sorted_tables:
  15. table.schema = None
  16. for col in table.c:
  17. if getattr(col.type, 'collation', None) is not None:
  18. col.type.collation = None
  19. if type(col.type) is sqlalchemy.dialects.mssql.base.BIT:
  20. col.type = Boolean()
  21. if type(col.type) is DATETIME:
  22. # print(table.name + '.' + col.name)
  23. col.server_default = None
  24. if type(col.type) is sqlalchemy.dialects.mssql.base.SMALLDATETIME:
  25. col.type = DATETIME()
  26. source_meta.create_all()