使用FastAPI与SQLAlchemy无法创建PostgreSQL表求助

问题描述 #

我正在学习使用FastAPI和SQLAlchemy构建API，但尝试运行脚本在PostgreSQL中创建表时毫无反应，没有任何效果。

以下是我的代码：

Settings类代码 #

class Settings(BaseSettings):
    DB_URL: str =  f'postgresql+asyncpg://{db_user}:{db_password}@{host}:{db_port}/{db_name}'
    DBbaseModel = declarative_base()

    class Config:
        case_sensitive = True


settings = Settings()

引擎与会话代码 #

engine : AsyncEngine = create_async_engine(settings.DB_URL)

Session: AsyncSession =  sessionmaker(
    autocommit=False, autoflush=False, expire_on_commit=False, class_=AsyncSession, bind=engine
)

创建表的脚本 #

async def create_tables() -> None:
    async with engine.begin() as connection:
        await connection.run_sync(settings.DBbaseModel.metadata.drop_all)
        await connection.run_sync(settings.DBbaseModel.metadata.create_all)


if __name__ == '__main__':
    import asyncio
    asyncio.run(create_tables())

补充模型代码 #

from core.configs import Settings
from sqlalchemy import Column, Integer, String

class UserModel(Settings.DBbaseModel):
    __tablename__ = 'users'
    id: Column(Integer, primary_key=True, autoincrement=True)
    fullname: Column(String)
    email: Column(String)
    password: Column(String)

核心问题分析 #

1. 模型字段定义错误：你的UserModel中，字段用了类型注解语法（id: Column(...)）而非赋值语法（id = Column(...)）。SQLAlchemy必须通过赋值把字段注册到模型元数据里，类型注解的写法不会被识别，导致元数据里没有任何表结构，create_all()自然不会执行任何操作。
2. Base实例定义位置不当：把declarative_base()放在Settings类内部，可能导致模型继承的Base实例和创建表时用的settings.DBbaseModel不是同一个（如果Settings被多次实例化），虽然这不是本次无反应的直接原因，但建议调整写法避免潜在问题。
3. DB_URL变量未定义：Settings里的DB_URL用到了db_user、db_password等变量，但没看到这些变量的定义，这会导致运行时错误，建议通过BaseSettings从环境变量读取，或者明确赋值。

修复后的代码示例 #

1. 修正Settings类 #

from pydantic import BaseSettings
from sqlalchemy.ext.declarative import declarative_base

# 先定义全局的Base实例，确保所有模型共享同一个元数据
Base = declarative_base()

class Settings(BaseSettings):
    DB_USER: str = "your_db_user"
    DB_PASSWORD: str = "your_db_password"
    DB_HOST: str = "localhost"
    DB_PORT: str = "5432"
    DB_NAME: str = "your_db_name"

    # 用属性动态生成DB_URL，避免硬编码
    @property
    def DB_URL(self):
        return f'postgresql+asyncpg://{self.DB_USER}:{self.DB_PASSWORD}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}'

    class Config:
        case_sensitive = True
        # 可选：从.env文件读取环境变量，方便不同环境配置
        env_file = ".env"

settings = Settings()

2. 修正UserModel #

from core.configs import Base, settings
from sqlalchemy import Column, Integer, String

class UserModel(Base):
    __tablename__ = 'users'
    # 用赋值语法定义字段
    id = Column(Integer, primary_key=True, autoincrement=True)
    fullname = Column(String(255))  # 建议指定字符串长度，符合数据库规范
    email = Column(String(255), unique=True)  # 邮箱设为唯一约束
    password = Column(String(255))

3. 引擎与会话代码（微调） #

from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from core.configs import settings

engine: AsyncEngine = create_async_engine(settings.DB_URL)

Session: AsyncSession = sessionmaker(
    autocommit=False, autoflush=False, expire_on_commit=False, class_=AsyncSession, bind=engine
)

4. 创建表脚本（修正元数据引用） #

from core.configs import engine, Base
import asyncio

async def create_tables() -> None:
    async with engine.begin() as connection:
        # 开发环境可选：先删除所有表
        await connection.run_sync(Base.metadata.drop_all)
        # 创建所有模型对应的表
        await connection.run_sync(Base.metadata.create_all)
    print("表创建完成！")

if __name__ == '__main__':
    asyncio.run(create_tables())

验证步骤 #

1. 确保PostgreSQL服务正在运行，且目标数据库your_db_name已手动创建（SQLAlchemy不会自动创建数据库）。
2. 确认依赖已安装：pip install asyncpg sqlalchemy pydantic fastapi。
3. 运行创建表脚本，查看控制台是否输出"表创建完成！"，再登录PostgreSQL检查users表是否存在。

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