Fastapi 从 postgreSQL 中的现有数据库表中读取

Question

我正在尝试创建一个 FAST Api，它从 PostgreSQL 数据库中现有的表中读取数据，但它给了我一个内部服务器错误。感谢您对代码可能存在的问题的指导

现有的表如下所示：

架构：testSchema
表：test_api

ID	电子邮件
1	测试@***.com
2	test2@***.com

engine = sqlalchemy.create_engine("my_database_connection")
Base = declarative_base()
database = databases.Database("my_database_connection")

metadata = sqlalchemy.MetaData()
metadata.reflect(bind=engine, schema='testSchema')
test_api_tb = metadata.tables['testSchema.test_api']
   


class testAPI(Base):
    __tablename__ = test_api_tb
    id = Column(Integer, primary_key=True)
    email = Column(String(256))

app = FastAPI()


@app.get("/testing_api/")
def read_users():
     query = test_api_tb.select()
     return  database.execute(query)

我从日志中得到的错误

RecursionError: maximum recursion depth exceeded in comparison

Answer 1

您能做的最好的事情就是阅读fastapi.tiangolo.com上的官方文档，它非常棒，并且非常详细地解释了所有基础知识。

SQL 关系数据库经常与 FastAPI 一起使用，并且在此处的文档中也提到，您可以找到有关如何将 postgresql 与 sqlalchemy 和 FastAPI 一起使用的分步教程。

有几个部分可以完成这项工作。第一部分是连接数据库：

engine = create_engine(my_database_connection)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

我们像您一样使用连接字符串创建引擎，然后我们需要创建一个会话才能连接到数据库。最后，我们创建一个基类，它将帮助我们创建模型和模式。

现在我们需要使用基类创建模型，就像上面所做的那样。我们需要确保它__tablename__与数据库中表的名称相同

class testAPIModel(Base):
    __tablename__ = "test_api"
    id = Column(Integer, primary_key=True)
    email = Column(String(256))

现在是主要部分。我们需要确保使用以下方法将数据库引擎绑定到基类

Base.metadata.create_all(bind=engine)

现在我们将创建一个函数，该函数将帮助我们创建数据库会话实例，并在完成查询后关闭连接。

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

现在我们可以创建 FastAPI 应用程序实例并从数据库中获取数据。

@app.get("/testing_api/")
def read_users(db:Session = Depends(get_db)):
     users = db.query(testAPIModel).all()
     return users

我们正在使用Depends(get_db)上面编写的函数注入数据库会话。

完整代码：

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

from fastapi import Depends, FastAPI
from sqlalchemy import Column, Integer, String

my_database_connection = "postgresql://user:password@server_ip/db_name"

engine = create_engine(my_database_connection)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class testAPIModel(Base):
    __tablename__ = "test_api"
    id = Column(Integer, primary_key=True)
    email = Column(String(256))


Base.metadata.create_all(bind=engine)
app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/testing_api/")
def read_users(db:Session = Depends(get_db)):
    users = db.query(testAPIModel).all()
    return users

祝你好运！