一、快速開始

1.1 簡介

Alembic 是一個基于 SQLAlchemy 的數據庫遷移工具，主要用于管理數據庫模式（Schema）的變更，例如新增表、修改字段、刪除索引等，確保數據庫結構與應用程序的 ORM 模型保持一致。

Alembic 通過版本控制來管理數據庫模式的變化，每次遷移都會生成一個唯一的版本文件，類似于 Git 的提交（commit）。

• 在線模式：直接連接數據庫，執行遷移。
• 離線模式：生成 SQL 語句，后續再執行。

1.2 安裝

uv add alembic

1.3 初始化

uv run alembic init alembic# 最后的alembic是項目的名字，可以修改，類似虛擬環境

1.4 alembic 目錄結構

project/
│── alembic/             # Alembic 遷移目錄
│   ├── versions/        # 遷移腳本存放目錄
│   ├── env.py           # 遷移腳本配置（數據庫連接、ORM 綁定等）
│   ├── script.py.mako   # 遷移腳本的模板
│── alembic.ini          # Alembic 配置文件

二、實戰

2.1 實戰簡介

使用 FastAPI + SQLModel + PostgreSQL + Alembic 測試 Alembic 數據庫遷移功能

2.2 項目初始化

• 創建測試文件夾

mkdir test-alembic

• 設置 uv 工程

cd test-alembic && uv init

如果你不知道 uv 是什么？推薦你看看博主的這篇文章喲~

Python 項目管理利器：uv 入門指南

• 增加項目依賴項

uv add fastapi sqlmodel psycopg2 alembic pydantic_core pydantic_settings

• 初始化 FastAPI 項目

from fastapi import FastAPIapp = FastAPI()

• 創建項目 settings.py 模塊

from pydantic import ConfigDict, computed_field, PostgresDsn
from pydantic_core import MultiHostUrl
from pydantic_settings import BaseSettingsclass Settings(BaseSettings):POSTGRES_SERVER: strPOSTGRES_PORT: int = 5432POSTGRES_USER: strPOSTGRES_PASSWORD: str = ""POSTGRES_DB: str = ""model_config = ConfigDict(env_file=".env")  # 使用 model_config 替代 class ConfigDict@computed_field  # type: ignore[prop-decorator]@propertydef SQLALCHEMY_DATABASE_URI(self) -> PostgresDsn:return MultiHostUrl.build(scheme="postgresql+psycopg2",username=self.POSTGRES_USER,password=self.POSTGRES_PASSWORD,host=self.POSTGRES_SERVER,port=self.POSTGRES_PORT,path=self.POSTGRES_DB,)settings = Settings()  # type: ignore

2.3 創建 models.py 模塊

import uuid
from datetime import datetimefrom sqlmodel import Field, SQLModelclass User(SQLModel, table=True):id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)user_account: str = Field(..., max_length=32, description="賬號")username: str = Field(default="", max_length=32, description="用戶昵稱", nullable=True)user_avatar: str = Field(default="", max_length=64, description="用戶頭像", nullable=True)user_profile: str = Field(default="", max_length=512, description="用戶簡介", nullable=True)user_role: str = Field(default="user", max_length=32, description="用戶角色：user/admin/ban")create_time: datetime = Field(default_factory=datetime.now, description="創建時間")update_time: datetime = Field(default_factory=datetime.now,sa_column_kwargs={"onupdate": datetime.now},description="更新時間",)is_delete: bool = Field(default=False, description="是否刪除", nullable=True)

2.4 初始化 Alembic

uv run alembic init alembic

2.5 連接 PG 數據庫

• 創建 .env 文件

POSTGRES_SERVER=localhost
POSTGRES_PORT=5432
POSTGRES_USER=admin
POSTGRES_PASSWORD=123456
POSTGRES_DB=test

假設你的 PG 數據庫設置是上面這些

• 修改 env.py 文件 如下

from logging.config import fileConfigfrom alembic import context
from sqlalchemy import engine_from_config, pool# 這是 Alembic 配置（Config）對象，它提供對當前使用的 .ini 配置文件中參數的訪問
config = context.config# 解釋 Python 日志配置文件
# 這行代碼的作用基本上是設置日志記錄器 loggers
fileConfig(config.config_file_name)# 在此添加你的模型的 MetaData 對象 以支持 自動生成（autogenerate） 遷移腳本
# User 必須存在，雖然看起來沒有使用到，但是這樣才會被SQLModel識別到！！！
from models import SQLModel, User
from settings import settingstarget_metadata = SQLModel.metadata# env.py 需要的其他配置值可以通過 config 獲取, 例如
# my_important_option = config.get_main_option("my_important_option")def get_url():return str(settings.SQLALCHEMY_DATABASE_URI)def run_migrations_offline():"""以離線模式運行遷移。在這種情況下，上下文（Context）僅使用數據庫 URL 進行配置，而不會創建引擎（Engine），盡管這里也可以使用引擎通過跳過引擎的創建，我們甚至不需要數據庫驅動（DBAPI）可用在此模式下，對 context.execute() 的調用會將指定的 SQL 語句直接輸出到遷移腳本中"""url = get_url()context.configure(url=url, target_metadata=target_metadata, literal_binds=True, compare_type=True)with context.begin_transaction():context.run_migrations()def run_migrations_online():"""以在線模式運行遷移。在這種情況下，我們需要創建一個引擎（Engine）并將連接（Connection）與上下文（Context）關聯"""configuration = config.get_section(config.config_ini_section)configuration["sqlalchemy.url"] = get_url()connectable = engine_from_config(configuration,prefix="sqlalchemy.",poolclass=pool.NullPool,)with connectable.connect() as connection:context.configure(connection=connection, target_metadata=target_metadata, compare_type=True)with context.begin_transaction():context.run_migrations()if context.is_offline_mode():run_migrations_offline()
else:run_migrations_online()

• 修改 script.py.mako 文件如下

"""${message}Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}"""
from typing import Sequence, Unionfrom alembic import op
import sqlalchemy as sa# 就是加了這一行
import sqlmodel.sql.sqltypes${imports if imports else ""}# revision identifiers, used by Alembic.
revision: str = ${repr(up_revision)}
down_revision: Union[str, None] = ${repr(down_revision)}
branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}def upgrade() -> None:"""Upgrade schema."""${upgrades if upgrades else "pass"}def downgrade() -> None:"""Downgrade schema."""${downgrades if downgrades else "pass"}

注意：這個很重要，不然生成的遷移腳本有問題！！！

2.6 實戰指令

• 創建遷移文件

uv run alembic revision --autogenerate -m "Initial migration"

• 執行遷移

uv run alembic upgrade head

• 回滾遷移

uv run alembic downgrade -1

• 查看當前版本

uv run alembic current

• 查看歷史版本

uv run alembic history