我正在学习如何使用fastapi和sql alchemy来构建一个api,但是当我试图让我的脚本在postgreSQL中创建表时,它就是不起作用,什么都没有发生。
下面是我的代码:
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)
2条答案
按热度按时间r3i60tvu1#
看起来像是使用
asyncpg
连接到PostgreSQL数据库,但实际上是使用run_sync
执行数据库命令。run_sync
旨在与同步函数一起使用,但您将其与异步函数一起使用。相反,您应该使用connection.run
来执行异步函数。像这样编辑代码:
rbpvctlc2#
解决!
问题是,我必须在模型中添加一行
__allow_unmapped__ = True
才能使其工作