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

Question

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

现有的表如下所示：

schema : testSchema
表：test_api

id	email
1	test@***.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

祝你好运！