我很抱歉是否曾有人问过这个问题，我尝试过使用google，并且已经阅读了很多关于stackoverflow的答案，但没有一个像我的那么具体。
所以，怎么了？好吧，我已经在python中建立了这个项目，我想实现的一件事就是将表从一个数据库复制到另一个数据库，其中包含所有约束，而没有约束。
至于我尝试过的事情，我已经尝试过使用etlalchemy，但不适用于Windows。我曾考虑过将表格保存在pandas数据框中并使用to_sql，但我不确定副本的保真度。
我曾考虑过sqlalchemy和alembic，但还没有真正找到有关该方法的教程。
关于我正在使用的数据库系统。我的计算机上安装了MySQL Workbench 8.0 CE和Oracle 11g。
基本上，我想使用Python 3.x将具有所有键，约束等的表从一个数据库复制或迁移到另一个数据库。谢谢您的时间。
这是我正在使用的代码：
from PyQt5 import QtCore, QtGui, QtWidgets, uic
from PyQt5.QtWidgets import QApplication, QMainWindow
from PyQt5.QtCore import QCoreApplication
from sqlalchemy import types, create_engine
import MySQLdb
import cx_Oracle
import pandas as pd
import sys
mySQLConn = MySQLdb.connect(host="localhost", # your host, usually localhost
user="user", # your username
passwd="pass", # your password
db="sakila") # name of the data base
#dsn_tns = cx_Oracle.makedsn('user', '1521', service_name='XE')
dsn_tns = cx_Oracle.makedsn('localhost', '1521', service_name='XE')
myOracleConn = cx_Oracle.connect(user=r'user', password='pass', dsn=dsn_tns)
#conn = create_engine('oracle+cx_oracle://user:pass@host:1521/?service_name=servicename')
class PandasModel(QtCore.QAbstractTableModel):
def __init__(self, df = pd.DataFrame(), parent=None):
QtCore.QAbstractTableModel.__init__(self, parent=parent)
self._df = df.copy()
def toDataFrame(self):
return self._df.copy()
def headerData(self, section, orientation, role=QtCore.Qt.DisplayRole):
if role != QtCore.Qt.DisplayRole:
return QtCore.QVariant()
if orientation == QtCore.Qt.Horizontal:
try:
return self._df.columns.tolist()[section]
except (IndexError, ):
return QtCore.QVariant()
elif orientation == QtCore.Qt.Vertical:
try:
# return self.df.index.tolist()
return self._df.index.tolist()[section]
except (IndexError, ):
return QtCore.QVariant()
def data(self, index, role=QtCore.Qt.DisplayRole):
if role != QtCore.Qt.DisplayRole:
return QtCore.QVariant()
if not index.isValid():
return QtCore.QVariant()
return QtCore.QVariant(str(self._df.ix[index.row(), index.column()]))
def setData(self, index, value, role):
row = self._df.index[index.row()]
col = self._df.columns[index.column()]
if hasattr(value, 'toPyObject'):
# PyQt4 gets a QVariant
value = value.toPyObject()
else:
# PySide gets an unicode
dtype = self._df[col].dtype
if dtype != object:
value = None if value == '' else dtype.type(value)
self._df.set_value(row, col, value)
return True
def rowCount(self, parent=QtCore.QModelIndex()):
return len(self._df.index)
def columnCount(self, parent=QtCore.QModelIndex()):
return len(self._df.columns)
def sort(self, column, order):
colname = self._df.columns.tolist()[column]
self.layoutAboutToBeChanged.emit()
self._df.sort_values(colname, ascending= order == QtCore.Qt.AscendingOrder, inplace=True)
self._df.reset_index(inplace=True, drop=True)
self.layoutChanged.emit()
#class usefulDatabaseFunctions():
# emptyDatabase = ""
#def getMySQLTables:
class usefulFunctions():
def connectToMySQL(self):
window.label.setText("Connected to MySQL")
window.setStyleSheet('QLabel#label {color: Green}')
#cur = db.cursor()
#cur.execute("SELECT * FROM actor")
#print(cur.fetchall())
def disconnectMySQL(self):
mySQLConn.close()
window.label.setText("Disconnected from MySQL")
window.setStyleSheet('QLabel#label {color: Red}')
print("Disconnected")
def connectToOracle(self):
window.label.setText("Connected to Oracle")
window.setStyleSheet('QLabel#label {color: Green}')
#cursorOracle = myOracleConn.cursor()
#cursorOracle.execute("SELECT * FROM countries")
#print(cursorOracle.fetchall())
def disconnectOracle(self):
window.label.setText("Disconnected from Oracle")
window.setStyleSheet('QLabel#label {color: Red}')
myOracleConn.close()
# def clearWork(self):
# window.queryTextEdit.setText("")
# window.resultsTextEdit.setText("")
# window.tableView.clearSpans()
# window.tableView.clear()
def executeMySQLUserQuery(self):
userQuery = window.queryTextEdit.toPlainText()
cursorMySQL = mySQLConn.cursor()
cursorMySQL.execute(userQuery)
print(cursorMySQL.fetchall())
df = pd.read_sql(userQuery, con = mySQLConn)
print(df.to_string())
#txtQueryResults = df.to_string()
txtQueryResults = str(cursorMySQL.fetchall())
window.resultsTextEdit.setPlainText(txtQueryResults)
model = PandasModel(df)
window.tableView.setModel(model)
def executeOracleUserQuery(self):
userQuery = window.queryTextEdit.toPlainText()
cursorOracle = myOracleConn.cursor()
cursorOracle.execute(userQuery)
print(cursorOracle.fetchall())
df = pd.read_sql(userQuery, con = myOracleConn)
print(df.to_string())
#txtQueryResults = df.to_string()
txtQueryResults = str(cursorOracle.fetchall())
window.resultsTextEdit.setText(txtQueryResults)
model = PandasModel(df)
window.tableView.setModel(model)
def clearWork(self): #Functie de stergere
window.queryTextEdit.setText("")
window.resultsTextEdit.setText("")
window.tableView.clearSpans()
# def migrateMySQLToOracle(self):
def exitProgram(self): #Functie de iesire
window.deleteLater()
window.close()
window.destroy()
uf = usefulFunctions
class Ui(QtWidgets.QMainWindow):
def __init__(self):
super(Ui, self).__init__() # Call the inherited classes __init__ method
uic.loadUi('Proiectv3.ui', self) # Load the .ui file
self.connectMySQLButton.clicked.connect(uf.connectToMySQL)
self.disconnectMySQLButton.clicked.connect(uf.disconnectMySQL)
self.connectOracleButton.clicked.connect(uf.connectToOracle)
self.disconnectOracleButton.clicked.connect(uf.disconnectOracle)
self.executeMySQLButton.clicked.connect(uf.executeMySQLUserQuery)
self.executeOracleButton.clicked.connect(uf.executeOracleUserQuery)
#self.transferButton.clicked.connect(uf.migrateMySQLToOracle)
self.clearButton.clicked.connect(uf.clearWork)
self.exitButton.clicked.connect(uf.exitProgram)
self.show() # Show the GUI
#app = QtWidgets.QApplication(sys.argv) # Create an instance of QtWidgets.QApplication
app = QCoreApplication.instance()
#app.setStyle('Fusion')
if app is None:
app = QApplication(sys.argv)
window = Ui() # Create an instance of our class
app.exec_() # Start the application
编辑1
这些是我尝试过的解决方案。
def migrateMySQLToOracle(self): #<-- Easieste solution and works.
dst = create_engine('oracle+cx_oracle://user:pass@localhost:1521/xe')
sql = 'SELECT * FROM actor'
df = pd.read_sql(sql, mySQLConn)
df.to_sql('actor', dst,if_exists = 'replace')
def migrateMySQLToOracle(self):
# create engine, reflect existing columns, and create table object for oldTable
#print('Test1')
srcEngine = create_engine('mysql+mysqldb://user:pass@localhost/classicmodels') # change this for your source database
srcEngine._metadata = MetaData(bind=srcEngine)
srcEngine._metadata.reflect(srcEngine) # get columns from existing table
srcTable = Table('offices', srcEngine._metadata)
# create engine and table object for newTable
#print('Test2')
destEngine = create_engine('oracle+cx_oracle://user:pass@localhost:1521/xe') # change this for your destination database
destEngine._metadata = MetaData(bind=destEngine)
destTable = Table('office', destEngine._metadata)
#print('Test3')
# copy schema and create newTable from oldTable
for column in srcTable.columns:
#print('Test4')
destTable.append_column(column.copy())
destTable.create()
def migrateMySQLToOracle(self):
src = create_engine('mysql+mysqldb://user:pass@localhost/classicmodels')
dst = create_engine('oracle+cx_oracle://user:pass@localhost:1521/xe')
meta = MetaData()
meta.reflect(bind=src)
tables = meta.tables
for tbl in tables:
data = src.execute(tables[tbl].select()).fetchmany()
if data:
dst.execute(tables[tbl].insert(), data)
def migrateMySQLToOracle(self):
print('Test1')
src = create_engine('mysql+mysqldb://user:pass@localhost/classicmodels')
dst = create_engine('oracle+cx_oracle://user:pass@localhost:1521/xe')
meta = MetaData()
tables = meta.tables;
print('Test2')
for tbl in tables:
print ('##################################')
print (tbl)
print ( tables[tbl].select())
data = src.execute(tables[tbl].select()).fetchall()
for a in data:
print(a)
if data:
print (tables[tbl].insert())
dst.execute( tables[tbl].insert(), data)
def migrateMySQLToOracle(self):
src = create_engine('mysql+mysqldb://user:pass@localhost/classicmodels')
dst = create_engine('oracle+cx_oracle://user:pass@localhost:1521/xe')
meta = MetaData()
meta.reflect(bind=src)
tables = meta.tables
for tbl in tables:
data = src.execute(tables[tbl].select()).fetchall()
if data:
dst.execute(tables[tbl].insert(), data)
def migrateMySQLToOracle(self):
old_base = automap_base()
old_engine = create_engine("mysql+mysqldb://user:pass@localhost/classicmodels", echo=True)
old_base.prepare(old_engine, reflect=True)
TableOld = old_base.classes.table_old
old_session = Session(old_engine)
new_base = automap_base()
new_engine = create_engine("oracle+cx_oracle://user:pass@localhost:1521/xe", echo=True)
new_base.prepare(new_engine, reflect=True)
TableNew = old_base.classes.table_new
new_session = Session(new_engine)
# here you can write your queries
old_table_results = old_session.query(TableOld).all()
new_data = []
for result in old_table_results:
new = TableNew()
new.id = result.id
new.name = result.name
new_data.append(new)
new_session.bulk_save_objects(new_data)
new_session.commit()
好，所以我已经回顾了一些我最初尝试过的解决方案。我得出的结论是，有两种方法可以将MySQL表复制到Oracle。
def migrateMySQLToOracle(self):
src = create_engine('mysql+mysqldb://<user>:<pass>@localhost/<database>')
dst = create_engine('oracle+cx_oracle://<user>:<pass>@localhost:port/<service>')
sql = 'SELECT * FROM <table>'
df = pd.read_sql(sql, src)
print(df)
df.to_sql('<Table name that will appear in Oracle>', dst,if_exists = 'replace')
第一个解决方案需要导入熊猫并使用熊猫
从技术上来讲，这是一个令人满意的方法，我注意到的唯一问题是，如果您想将此解决方案应用于从Oracle复制表到MySQL的过程中，它将遇到来自Oracle的NUMBER数据类型的麻烦，而python控制台将输出错误。否则，它确实可以工作，只要确保表中没有NUMBER数据类型列即可。
https://www.slideshare.net/Stiivi/python-business-intelligence-pydata-2012-talk
使用此链接，从幻灯片35开始。它需要使用sqlalchemy