跳到内容

使用 FastAPI 的简单英雄 API

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

安装 FastAPI

第一步是安装 FastAPI。

FastAPI 是创建 web API 的框架。

但是我们还需要另一种类型的程序来运行它，它被称为“服务器”。我们将使用 Uvicorn。我们将安装带有其标准依赖项的 Uvicorn。

然后安装 FastAPI。

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

$ pip install fastapi "uvicorn[standard]"

---> 100%

SQLModel 代码 - 模型、Engine

现在让我们从 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_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 First Steps 文档。

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

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

# 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

这非常简单直接。

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

每个请求一个 Session

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

在这里，它更加明显。

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

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

但是我们永远不想在不同的请求之间共享同一个 session。

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

在以后的示例中，我们将使用 FastAPI 依赖项 来获取 session，使其能够与其他依赖项共享，并能够在测试期间替换它。 🤓

运行 FastAPI 应用程序

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

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

然后使用 Uvicorn 运行它

$ uvicorn main:app

<span style="color: green;">INFO</span>:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
<span style="color: green;">INFO</span>:     Started reloader process [28720]
<span style="color: green;">INFO</span>:     Started server process [28722]
<span style="color: green;">INFO</span>:     Waiting for application startup.
<span style="color: green;">INFO</span>:     Application startup complete.

信息

命令 uvicorn main:app 指的是

  • main：文件 main.py（Python “模块”）。
  • app：在 main.py 中使用行 app = FastAPI() 创建的对象。

Uvicorn --reload

在开发期间（并且仅在开发期间），你还可以向 Uvicorn 添加选项 --reload。

它会在你每次更改代码时重启服务器，这样你就可以更快地开发。 🤓

$ uvicorn main:app --reload

<span style="color: green;">INFO</span>:     Will watch for changes in these directories: ['/home/user/code/sqlmodel-tutorial']
<span style="color: green;">INFO</span>:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
<span style="color: green;">INFO</span>:     Started reloader process [28720]
<span style="color: green;">INFO</span>:     Started server process [28722]
<span style="color: green;">INFO</span>:     Waiting for application startup.
<span style="color: green;">INFO</span>:     Application startup complete.

只需记住永远不要在生产环境中使用 --reload，因为它会消耗比必要更多的资源，更容易出错等等。

检查 API 文档 UI

现在你可以转到浏览器中的 URL http://127.0.0.1:8000。我们没有为根路径 / 创建路径操作，所以单独的 URL 只会显示“Not Found”错误...... “Not Found”错误是由你的 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 来终止 Uvicorn 服务器。

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

DB Browser for SQLite showing the heroes

总结

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

我们可以改进和扩展一些东西。例如，我们希望数据库决定每个新英雄的 ID，我们不想允许用户发送它。

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