尝试将数据库迁移到我的早期项目中。虽然构建所有内容的代码很简陋,但db结构和数据本身相当合理。我可能想出一个迁移所有东西的转储方法(pgdump等),但是我仍然需要学习很多这些东西,所以我宁愿通过一步一步的方式来获得经验。
来源:一个~~ 1gb的sqlite数据库文件
目的地:运行Postgres v9.6的Google CloudSQL
已经在cloud db中创建了表,使用与sqlite db相同的模式和表名。不担心模式执行错误,因为我还没有在云中定义外键。
计划:为每个数据库创建不同的并发SQLAlchemy连接,然后读取sqlite - >写入CloudSQL。使用SQLAlchemy返回并定义每个表的数据结构。来自models.py的片段:
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base, declared_attr
Base = declarative_base()
class PublicMixin(object):
@declared_attr
def __tablename__(cls):
return cls.__name__.lower()
class Category(PublicMixin, Base):
id = Column(Integer, primary_key=True)
name = Column(String)
class Player(PublicMixin, Base):
id = Column(Integer, primary_key=True)
name = Column(String)
username = Column(String)
notes = Column(String)
[...]
然后我将此文件复制为models_lite.py,这样我就可以导入每个模型而不会产生干扰。这是我试图运行的migration.py文件作为概念证明:
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Base, Category, Player
from models_sqlite import Base as Base_lite, Category as Category_lite, Player as Player_lite
# SQLite db
engine_lite = create_engine('sqlite:///mydb.sqlite')
Base_lite.metadata.bind = engine_lite
LiteSession = sessionmaker()
LiteSession.bind = engine_lite
session_lite = LiteSession()
# CloudSQL, via local proxy
engine_cloud = create_engine('postgresql+psycopg2://USER:PW@/DBNAME?host=/cloudsql/INSTANCE')
Base.metadata.bind = engine_cloud
CloudSession = sessionmaker()
CloudSession.bind = engine_cloud
session_cloud = CloudSession()
category_lite = session_lite.query(Category_lite).all()
category_cloud = Category()
for c in category_lite:
category_cloud = c
session_cloud.add(category_cloud)
session_cloud.commit()
运行此命令会产生以下错误:
File "postgres migration.py", line 68, in <module>
session_cloud.add(category_cloud)
[...]
sqlalchemy.exc.InvalidRequestError: Object '<Category at 0x11141b908>' is already attached to session '1' (this is '2')
明确地设置for
循环中的每一列是有效的(即:category_cloud.id = c.id
),但必须有一种方法可以避免必须为每个表中的每一列执行此操作。我该怎么做呢?
使用sqlalchemy核心而不是orm更容易实现这种数据传输操作。将数据库数据映射到此处的对象没有任何好处,如果它们只是立即写入另一个数据库,它只会增加复杂性并减慢速度。以下代码将遍历Base中的每个表,选择sqlite数据库中的所有列,并将它们一次写入云数据库。
from sqlalchemy import create_engine, select
from models import Base
engine_lite = create_engine('sqlite:///mydb.sqlite')
engine_cloud = create_engine('postgresql+psycopg2://USER:PW@/DBNAME?host=/cloudsql/INSTANCE')
with engine_lite.connect() as conn_lite:
with engine_cloud.connect() as conn_cloud:
for table in Base.metadata.sorted_tables:
data = [dict(row) for row in conn_lite.execute(select(table.c))]
conn_cloud.execute(table.insert().values(data))