多数据库迁移的艺术：Alembic在复杂环境中的精妙应用
title: 多数据库迁移的艺术：Alembic在复杂环境中的精妙应用
project/
├── alembic/
│ ├── versions/
│ │ ├── db1/
│ │ └── db2/
│ └── env.py
├── alembic.ini
└── app/
└── models.py
[alembic]
script_location = alembic
[db1]
sqlalchemy.url = postgresql://user:pass@localhost/db1
[db2]
sqlalchemy.url = mysql://user:pass@localhost/db2
from alembic import context
def run_migrations(engine_name):
config = context.config
section = config.get_section(engine_name)
url = section["sqlalchemy.url"]
engine = create_engine(url)
with engine.connect() as connection:
context.configure(
connection=connection,
target_metadata=get_metadata(engine_name),
version_table=f'alembic_version_{engine_name}'
)
with context.begin_transaction():
context.run_migrations(engine_name)
if context.is_offline_mode():
run_migrations('db1')
run_migrations('db2')
else:
for engine_name in ['db1', 'db2']:
run_migrations(engine_name)
from sqlalchemy.ext.declarative import declarative_base
from pydantic import BaseModel
class UserBase(BaseModel):
email: str
is_active: bool = True
class UserCreate(UserBase):
password: str
class UserDB(UserBase):
id: int
Db1Base = declarative_base()
Db2Base = declarative_base()
class Db1User(Db1Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
email = Column(String)
password_hash = Column(String)
is_active = Column(Boolean)
class Db2Log(Db2Base):
__tablename__ = "logs"
id = Column(Integer, primary_key=True)
action = Column(String)
user_id = Column(Integer)
alembic -n db1 revision --autogenerate -m "add users table"
alembic -n db2 revision --autogenerate -m "add logs table"
alembic -n db1 upgrade head
alembic -n db2 upgrade head
alembic -n db1 history --verbose
alembic -n db2 history --verbose
alembic -n db1 revision --autogenerate --version-path=alembic/versions/db1
alembic -n db1 stamp head
[db3]
sqlalchemy.url = sqlite:///db3.sqlite
Db1Base.metadata.create_all(engine) # 在应用启动时执行
评论
发表评论