FastAPI启动时自动初始化PostgreSQL管理员账号方案咨询

核心思路 #

在FastAPI应用启动阶段，通过startup事件触发管理员账号检查逻辑：从.env读取预设的管理员账号信息，使用SQLAlchemy查询数据库users表，若目标账号不存在则自动创建（包含密码哈希处理）。

依赖准备 #

先安装所需依赖包：

pip install fastapi uvicorn sqlalchemy psycopg2-binary python-dotenv passlib bcrypt

完整实现代码 #

1. .env配置文件

在项目根目录创建.env，填入数据库连接信息和管理员账号：

# 数据库连接地址
DATABASE_URL=postgresql://your_db_user:your_db_password@localhost/your_db_name
# 管理员账号信息
ADMIN_USERNAME=admin
ADMIN_PASSWORD=your_secure_admin_password

2. FastAPI应用代码

from fastapi import FastAPI
from sqlalchemy import create_engine, Column, Integer, String, DateTime, func, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from dotenv import load_dotenv
from passlib.context import CryptContext
import os
from contextlib import contextmanager

# 加载环境变量
load_dotenv()

# 密码哈希配置
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# 数据库初始化
DATABASE_URL = os.getenv("DATABASE_URL")
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# 用户模型（保留你的定义，补充必要约束）
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    username = Column(String, unique=True, nullable=False)  # 新增唯一约束与非空约束
    password_hash = Column(String, nullable=False)
    role = Column(String, default='admin')
    created_at = Column(DateTime, default=func.now())
    created_by = Column(Integer, ForeignKey('users.id'), default=1)
    modified_at = Column(DateTime, default=func.now(), onupdate=func.now())
    modified_by = Column(Integer, ForeignKey('users.id'), default=1)

# 创建数据库表（首次运行自动生成）
Base.metadata.create_all(bind=engine)

# 数据库会话上下文管理器
@contextmanager
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 管理员账号初始化逻辑
def init_admin_account():
    admin_username = os.getenv("ADMIN_USERNAME")
    admin_password = os.getenv("ADMIN_PASSWORD")
    
    if not admin_username or not admin_password:
        raise ValueError(".env文件中未配置ADMIN_USERNAME或ADMIN_PASSWORD")
    
    with get_db() as db:
        # 检查账号是否已存在
        existing_user = db.query(User).filter(User.username == admin_username).first()
        if not existing_user:
            # 生成密码哈希
            hashed_pwd = pwd_context.hash(admin_password)
            # 创建管理员用户（created_by默认1，插入后用户ID即为1，满足外键约束）
            admin_user = User(
                username=admin_username,
                password_hash=hashed_pwd
            )
            db.add(admin_user)
            db.commit()
            print(f"✅ 管理员账号 {admin_username} 已创建")
        else:
            print(f"ℹ️ 管理员账号 {admin_username} 已存在")

# FastAPI应用实例
app = FastAPI()

# 启动时触发初始化
@app.on_event("startup")
def startup():
    init_admin_account()

关键说明 #

• 启动事件触发：通过@app.on_event("startup")装饰器，确保初始化逻辑在应用启动时仅执行一次，后续重启若账号已存在则跳过。
• 密码安全：使用passlib的bcrypt算法哈希密码，避免明文存储。
• 外键约束兼容：第一个管理员用户的created_by默认值为1，而用户插入后ID自动生成为1，刚好满足外键指向自身的约束。
• 表自动创建：通过Base.metadata.create_all(bind=engine)首次运行时自动生成users表，无需手动建表。

内容的提问来源于stack exchange，提问作者Dmitry Mogilnikov