使用 FastAPI 的简单英雄 API¶

让我们从使用 FastAPI 构建一个简单的英雄 Web API 开始。✨

安装 FastAPI¶

第一步是安装 FastAPI。

FastAPI 是创建 Web API 的框架。

请确保您创建一个虚拟环境，激活它，然后安装它们，例如使用

$ pip install fastapi "uvicorn[standard]"

---> 100%

SQLModel 代码 - 模型、引擎¶

现在让我们开始 SQLModel 代码。

我们将从 最简单的版本 开始，只有英雄（还没有团队）。

这与我们迄今为止在前面的示例中看到的几乎是相同的代码

Python 3.10+

# Code above omitted 👆

from sqlmodel import Field, Session, SQLModel, create_engine, select

# Code here omitted 👈

class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

# Code below omitted 👇

👀 完整文件预览

Python 3.10+

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/")
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


@app.get("/heroes/")
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

🤓 其他版本和变体

Python 3.8+

from typing import Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/")
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


@app.get("/heroes/")
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

与我们之前使用的代码相比，这里只有一个变化，即 connect_args 中的 check_same_thread。

这是一个 SQLAlchemy 传递给负责与数据库通信的底层库的配置。

check_same_thread 默认设置为 True，以防止在某些简单情况下的误用。

但在这里，我们将确保不在一个请求中共享相同的会话，这是防止该配置存在问题的实际 最安全的方法。

我们还需要禁用它，因为在 FastAPI 中，每个请求都可能由多个交互线程处理。

信息

这些信息现在已经足够了，您可以在 FastAPI 关于 async 和 await 的文档中阅读更多信息。

重点是，通过确保您 不共享 相同的会话与多个请求，代码已经是安全的。

FastAPI 应用程序¶

下一步是创建 FastAPI 应用程序。

我们将从 fastapi 导入 FastAPI 类。

然后创建一个 app 对象，它是 FastAPI 类的一个实例

Python 3.10+

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select

# Code here omitted 👈

app = FastAPI()

# Code below omitted 👇

👀 完整文件预览

Python 3.10+

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/")
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


@app.get("/heroes/")
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

🤓 其他版本和变体

Python 3.8+

from typing import Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/")
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


@app.get("/heroes/")
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

在 `startup` 时创建数据库和表¶

我们希望确保一旦应用程序开始运行，就会调用 create_db_and_tables 函数。以创建数据库和表。

这应该在启动时只调用一次，而不是在每个请求之前，所以我们将其放在处理 "startup" 事件的函数中

Python 3.10+

# Code above omitted 👆

app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()

# Code below omitted 👇

👀 完整文件预览

Python 3.10+

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/")
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


@app.get("/heroes/")
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

🤓 其他版本和变体

Python 3.8+

from typing import Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/")
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


@app.get("/heroes/")
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

创建英雄路径操作¶

信息

如果您需要复习什么是 路径操作（一个具有特定 HTTP 操作的端点）以及如何在 FastAPI 中使用它，请查看 FastAPI 快速入门文档。

让我们创建 路径操作 代码来创建一个新英雄。

当用户向 /heroes/ 路径发送带有 POST 操作的请求时，它将被调用

Python 3.10+

# Code above omitted 👆

app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/")
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero

# Code below omitted 👇

👀 完整文件预览

Python 3.10+

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/")
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


@app.get("/heroes/")
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

🤓 其他版本和变体

Python 3.8+

from typing import Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/")
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


@app.get("/heroes/")
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

信息

如果您需要复习其中一些概念，请查看 FastAPI 文档

SQLModel 优势¶

这就是我们的 SQLModel 类模型同时作为 SQLAlchemy 模型和 Pydantic 模型时的优势所在。✨

在这里，我们使用相同的类模型来定义将由我们的 API 接收的 请求正文。

因为 FastAPI 基于 Pydantic，它将使用相同的模型（Pydantic 部分）进行自动数据验证和从 JSON 请求到 Hero 类实际实例的对象的转换。

然后，因为这个相同的 SQLModel 对象不仅是一个 Pydantic 模型实例，而且还是一个 SQLAlchemy 模型实例，我们可以直接在会话中使用它来在数据库中创建行。

所以我们可以使用直观的标准 Python 类型注解，并且我们不必为数据库模型和 API 数据模型重复大量代码。🎉

提示

我们稍后会进一步改进这一点，但目前，它已经展示了 SQLModel 类同时作为 SQLAlchemy 模型和 Pydantic 模型的强大功能。

读取英雄路径操作¶

现在让我们添加另一个 路径操作 来读取所有英雄

Python 3.10+

# Code above omitted 👆

app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/")
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


@app.get("/heroes/")
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

👀 完整文件预览

Python 3.10+

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/")
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


@app.get("/heroes/")
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

🤓 其他版本和变体

Python 3.8+

from typing import Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/")
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


@app.get("/heroes/")
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

这非常简单。

当客户端向 /heroes/ 路径发送带有 GET HTTP 操作的请求时，我们运行此函数，该函数从数据库获取英雄并返回它们。

每个请求一个会话¶

还记得我们应该为每组操作使用一个 SQLModel 会话，如果我们需要其他不相关的操作，我们应该使用不同的会话吗？

这里更明显。

在大多数情况下，我们通常应该 每个请求一个会话。

在某些孤立的情况下，我们希望在内部拥有新的会话，因此 每个请求多个会话。

但我们 绝不想在不同请求之间共享同一个会话。

在这个简单的例子中，我们只是在 路径操作函数 中手动创建新的会话。

在未来的示例中，我们将使用 FastAPI 依赖项来获取会话，从而能够与其他依赖项共享它，并且能够在测试期间替换它。🤓

在开发模式下运行 FastAPI 服务器¶

现在我们准备运行 FastAPI 应用程序。

将所有代码放入一个名为 main.py 的文件中。

然后使用 fastapi CLI 在开发模式下运行它

$ fastapi dev main.py

<span style="color: green;">INFO</span>:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

信息

fastapi 命令在底层使用 Uvicorn。

当您使用 fastapi dev 时，它会启动 Uvicorn 并带有每次更改代码时自动重新加载的选项，这样您就可以更快地开发。🤓

在生产模式下运行 FastAPI 服务器¶

开发模式不应用于生产，因为它默认包含自动重新加载，会消耗比必要更多的资源，并且更容易出错等。

对于生产环境，请使用 fastapi run 而不是 fastapi dev

$ fastapi run main.py

<span style="color: green;">INFO</span>:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

检查 API 文档 UI¶

现在您可以在浏览器中访问该 URL http://127.0.0.1:8000。我们没有为根路径 / 创建 路径操作，所以单独的该 URL 只会显示“未找到”错误……该“未找到”错误是由您的 FastAPI 应用程序生成的。

但是您可以访问路径 /docs 处的 自动生成的交互式 API 文档：http://127.0.0.1:8000/docs。✨

您会看到这个 自动 API 文档 UI 具有我们上面定义的路径及其操作，并且它已经知道 路径操作 将接收的数据的形状

Interactive API docs UI

使用 API¶

您可以点击 试一试 按钮并发送一些请求，使用 创建英雄 路径操作 来创建一些英雄。

然后您可以使用 读取英雄 路径操作 将它们取回

Interactive API docs UI reading heroes

检查数据库¶

现在您可以回到终端并按 Ctrl+C 来终止该服务器程序。

然后，您可以打开 DB Browser for SQLite 并检查数据库，以探索数据并确认它确实保存了英雄。🎉

DB Browser for SQLite showing the heroes

回顾¶

做得好！这已经是一个 FastAPI Web API 应用程序，用于与英雄数据库交互。🎉

有几件事我们可以改进和扩展。例如，我们希望数据库决定每个新英雄的 ID，我们不希望允许用户发送它。

我们将在接下来的章节中进行所有这些改进。🚀

使用 FastAPI 的简单英雄 API¶

安装 FastAPI¶

SQLModel 代码 - 模型、引擎¶

FastAPI 应用程序¶

在 startup 时创建数据库和表¶

创建英雄 路径操作¶

SQLModel 优势¶

读取英雄 路径操作¶

每个请求一个会话¶

在开发模式下运行 FastAPI 服务器¶

在生产模式下运行 FastAPI 服务器¶

检查 API 文档 UI¶

使用 API¶

检查数据库¶

回顾¶

在 `startup` 时创建数据库和表¶

创建英雄路径操作¶

读取英雄路径操作¶