# mainsever

**Repository Path**: zhxm_master/mainsever

## Basic Information

- **Project Name**: mainsever
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2025-12-01
- **Last Updated**: 2026-01-22

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

一步步集成 Alembic
安装依赖
```
pip install -r requirements.txt
```

初始化 Alembic 配置

在项目根目录执行：
```
alembic init alembic
```

它会生成结构：

```
alembic/
│
├── versions/          # 存放每次迁移脚本
├── env.py             # 环境配置文件
└── script.py.mako     # 迁移模板
alembic.ini            # Alembic 主配置文件
```

修改 alembic.ini

找到：
```
sqlalchemy.url = driver://user:pass@localhost/dbname
```
修改成，成动态加载（我们从 FastAPI 配置导入）：
```
sqlalchemy.url = sqlite:///placeholder.db
```

然后我们在 env.py 中动态设置连接。


修改 alembic/env.py 以支持 async + FastAPI config

把内容改成以下完整版本 
```
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from sqlalchemy.ext.asyncio import AsyncEngine
from alembic import context

from app.core.config import settings
from app.core.database import Base

# 导入所有模型
from app.models import user  # 导入模型以便 Alembic 能发现表结构

config = context.config
fileConfig(config.config_file_name)

target_metadata = Base.metadata

def get_url():
    return settings.SQLALCHEMY_DATABASE_URL

def run_migrations_offline():
    """Offline mode"""
    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 do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata)

    with context.begin_transaction():
        context.run_migrations()

async def run_migrations_online():
    """Async mode"""
    connectable = AsyncEngine(
        engine_from_config(
            config.get_section(config.config_ini_section),
            prefix="sqlalchemy.",
            poolclass=pool.NullPool,
            future=True,
            url=get_url(),
        )
    )

    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await connectable.dispose()

if context.is_offline_mode():
    run_migrations_offline()
else:
    import asyncio
    asyncio.run(run_migrations_online())

```


生成迁移脚本
检测模型变更并生成迁移文件：
```
alembic revision --autogenerate -m "init users table"
```

执行迁移（创建表）
```
alembic upgrade head
``

常用命令速查

| 功能          | 命令                                        |
| ----------- | ----------------------------------------- |
| 初始化 Alembic | `alembic init alembic`                    |
| 生成迁移脚本      | `alembic revision --autogenerate -m "描述"` |
| 执行迁移        | `alembic upgrade head`                    |
| 回滚到上个版本     | `alembic downgrade -1`                    |
| 查看迁移历史      | `alembic history`                         |
| 查看当前版本      | `alembic current`                         |
`