跳到内容

使用 FastAPI 构建简单的英雄 API

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

安装 FastAPI

第一步是安装 FastAPI。

FastAPI 是创建 Web API 的框架。

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

$ pip install fastapi "uvicorn[standard]"

---> 100%

SQLModel 代码 - 模型，引擎

现在让我们开始编写 SQLModel 代码。

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

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

# 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 👇
👀 完整文件预览 
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
🤓 其他版本和变体 
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，以防止在某些简单情况下被误用。

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

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

信息

目前这些信息就足够了，您可以在 FastAPI 文档中了解更多关于 async 和 await 的信息。

重点是，通过确保您不与多个请求共享同一个 session，代码已经很安全了。

FastAPI 应用

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

我们将从 fastapi 导入 FastAPI 类。

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

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

# Code here omitted 👈

app = FastAPI()

# Code below omitted 👇
👀 完整文件预览 
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
🤓 其他版本和变体 
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" 事件的函数中

# Code above omitted 👆

app = FastAPI()


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

# Code below omitted 👇
👀 完整文件预览 
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
🤓 其他版本和变体 
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 操作的请求时，它将被调用

# 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 👇
👀 完整文件预览 
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
🤓 其他版本和变体 
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 模型实例，我们可以直接在 session 中使用它来在数据库中创建行。

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

提示

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

读取英雄 路径操作

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

# 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
👀 完整文件预览 
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
🤓 其他版本和变体 
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 依赖项来获取 session，以便能够与其他依赖项共享它，并在测试期间能够替换它。🤓

在开发模式下运行 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

您实际上可以点击 Try it out 按钮并发送一些请求，使用创建英雄 路径操作 来创建一些英雄。

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

Interactive API docs UI reading heroes

检查数据库

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

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

DB Browser for SQLite showing the heroes

回顾

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

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

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