fastapi教程（一）：初识 fastapi

FastAPI 是一个用于构建 API 的现代、快速（高性能）的 web 框架，使用 Python 并基于标准的 Python 类型提示。

关键特性:

  • 快速：可与 NodeJS 和 Go 并肩的极高性能（归功于 Starlette 和 Pydantic）。最快的 Python web 框架之一。
  • 高效编码：提高功能开发速度约 200％ 至 300％。
  • 更少 bug：减少约 40％ 的人为（开发者）导致错误。
  • 智能：极佳的编辑器支持。处处皆可自动补全，减少调试时间。
  • 简单：设计的易于使用和学习，阅读文档的时间更短。
  • 简短：使代码重复最小化。通过不同的参数声明实现丰富功能。bug 更少。
  • 健壮：生产可用级别的代码。还有自动生成的交互式文档。
  • 标准化：基于（并完全兼容）API 的相关开放标准：OpenAPI (以前被称为 Swagger) 和 JSON Schema。

简单来体验一下：
环境：

  • Python 3.11
  • Windows 10
  • fastapi 0.68.0

1，安装 fastapi

pip install fastapi

2，安装一个 ASGI 服务器

pip install "uvicorn[standard]"
  • 生产环境可以使用 Uvicorn 或者 Hypercorn。

ASGI(Asynchronous Server Gateway Interface)是 Python 异步服务器网关接口的缩写。
它定义了异步编程风格的 Python Web 服务器与 Web 服务器和应用程序之间的标准接口。
ASGI 旨在解决传统 Python Web 服务器(如WSGI)在高并发和长连接场景下的不足，使用异步编程模型来提高服务器的并发处理能力。
ASGI服务器具有以下几个关键特点:

  • 异步和事件驱动：ASGI 服务器采用异步非阻塞的编程模型，可以高效处理大量并发连接，避免进程/线程阻塞。它使用事件循环机制，在有事件到来时处理，否则释放资源做其他事情。
  • 支持长连接和WebSocket：ASGI 服务器本身支持长连接和 WebSocket 协议，可用于构建实时通信应用，如在线聊天室、协作平台等。
    标准统一接口：ASGI 定义了统一的服务器端接口规范，使Web服务器、中间件和应用程序之间松散耦合，易于拓展和维护。
    高性能：相比传统 WSGI，ASGI 编程模型使用异步协程可以大幅降低上下文切换和内存开销，从而提高并发性能。

目前支持 ASGI 的服务器有 Uvicorn、Daphne、Hypercorn 等，应用框架包括 Django Channels、FastAPI 等。ASGI 正在逐渐取代 WSGI 成为 Python Web 应用开发的新标准，特别适合于需要高并发、实时双向通信的应用场景。

3，创建一个 main.py 文件

from typing import Union

from fastapi import FastAPI

# 创建 FastAPI 实例
app = FastAPI()

# 定义路由，处理 get 请求
@app.get("/")
# 定义路由处理函数
async def read_root():
    return {"Hello": "World"}

# 定义路由，处理 get 请求，接收一个路径参数 item_id
@app.get("/items/{item_id}")
async def read_item(
        item_id: int,
        q: str = 'default',
):
    return {"item_id": item_id, "q": q}

4，运行项目

$ uvicorn main:app --reload
INFO:     Will watch for changes in these directories: ['I:\\Python\\Web\\FastAPI\\fastApiProject1']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [14776] using WatchFiles
INFO:     Started server process [24944]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
  • main：main.py 文件（一个 Python “模块”）。
  • app：在 main.py 文件中通过 app = FastAPI() 创建的对象。
  • –reload：让服务器在更新代码后重新启动。仅在开发时使用该选项。

5，浏览器访问路由

访问 http://127.0.0.1:8000/ 可以在页面看到路由处理函数返回的数据：
在这里插入图片描述
访问 http://127.0.0.1:8000/items/5?q=somequery 可以在页面看到路由处理函数返回的数据：
在这里插入图片描述

6，体验类型声明的优势

浏览器访问 http://127.0.0.1:8000/docs 可以到达 API 界面：
在这里插入图片描述

但是国内可能打不开。
fastapi访问/docs和/redoc接口文档显示空白或无法加载
FastAPI 自动文档Swagger UI 打不开。显示空白

打开这个页面，我们可以看到非常详细的 API 描述：
在这里插入图片描述

  • item_id：integer类型，是一个路径参数，而且是必需的。
  • q：string 类型，是一个查询参数，而且不是必需的。

这是怎么做到的？回到代码：

@app.get("/items/{item_id}")
def read_item(
        item_id: int,
        q: str = 'default',
):
    return {"item_id": item_id, "q": q}

路由装饰器中定义了路径 /items/{item_id}，路径参数 item_id 在路由处理函数中明确定义为 int 数据类型，且未指定默认值，路径参数是用户必须传递的。

路由处理函数中，q 明确定义为 str 数据类型，且赋予了一个默认值 default，这种不是路径参数的路由处理函数参数就是查询参数。

  • 把默认值设为 None 即可声明可选的查询参数

最重要的特征就是，我们为参数添加了类型声明，这样做有非常多好处：

  1. 由 FastAPI 通过类型声明自动将解析请求中的字符串转为声明的数据类型：
    在这里插入图片描述
    2，更方便地利用编辑器的自动提示：
    在这里插入图片描述

7，借助 Pydantic 完成数据验证

多数情况我们会用 POST（最常用）、PUT 操作来发送数据，具体就是将数据放在这些请求的请求体中。
既然路径参数和查询参数都可以通过定义数据类型来完成默认的类型转化和校验，那么作为传述数据主要载体的请求体，自然也支持数据类型的定义带来的好处，响应体也是如此！

请求体和响应体的数据定义需要使用 pandatic，这是一个流行的数据验证工具。使用 Pydantic 模型声明请求体，能充分利用它的功能和优点：

  • 以 JSON 形式读取请求体。（在必要时）把请求体转换为对应的类型
  • 校验数据：数据无效时返回错误信息，并指出错误数据的确切位置和内容
  • 把接收的数据赋值给参数 item：把函数中请求体参数的类型声明为 Item，还能获得代码补全等编辑器支持
  • 为模型生成 JSON Schema，在项目中所需的位置使用
  • 用于 API 文档 UI
from fastapi import FastAPI
from pydantic import BaseModel


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None


app = FastAPI()


@app.post("/items/")
async def create_item(item: Item):
    return item
  • 修改代码并保存后，服务器将会自动重载（因为在上面的步骤中你向 uvicorn 命令添加了 --reload 选项来启动服务）

我们通过使继承 pydantic 的 BaseModel 类来定义了一个类，通常会将它称为模型，在这个模型中我们定义了一些字段及其数据类型，然后我们用这个模型定义一个查询参数的类型。

  • 使用Pydantic 模型的参数，是请求体。
  • 与声明查询参数一样，包含默认值的模型属性是可选的，否则就是必选的。默认值为 None 的模型属性也是可选的。

上述模型声明如下 JSON 对象（即 Python 字典）：

{
    "name": "Foo",
    "description": "An optional description",
    "price": 45.2,
    "tax": 3.5
}

运行项目，使用接口调试工具进行调试：
在这里插入图片描述
如果请求体数据必填项缺失：

发送：
{
    "name": "Foo",
    "description": "An optional description",
    "tax": 3.5
}
返回：
{
	"detail": [
		{
			"type": "missing",
			"loc": [
				"body",
				"price"
			],
			"msg": "Field required",
			"input": {
				"name": "Foo",
				"description": "An optional description",
				"tax": 3.5
			}
		}
	]
}

如果数据类型错误：

发送：
{
    "name": "Foo",
    "description": "An optional description",
    "price": "error type",
    "tax": 3.5
}
返回：
{
	"detail": [
		{
			"type": "float_parsing",
			"loc": [
				"body",
				"price"
			],
			"msg": "Input should be a valid number, unable to parse string as a number",
			"input": "error type"
		}
	]
}

这就是 pandatic 实现请求体数据校验的最大好处！

接下来看看响应体：可以在任意的路径操作中使用 response_model 参数来声明用于响应的模型。

from typing import Any

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()


class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: str | None = None


class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: str | None = None


@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn) -> Any:
    return user


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

即便我们的路径操作函数将会返回包含密码的相同输入用户，但我们已经将 response_model 声明为了不包含密码的 UserOut 模型，FastAPI 将会负责过滤掉未在输出模型中声明的所有数据。

发送：
{
    "username": "Jonas Vingegaard",
    "password": "123456",
    "email": "test@test.com",
    "full_name": "Jonas Vingegaard Rasmussen"
}
返回：
{
	"username": "Jonas Vingegaard",
	"email": "test@test.com",
	"full_name": "Jonas Vingegaard Rasmussen"
}

发送：
{
    "username": "Jonas Vingegaard",
    "password": "123456",
    "email": "test@test.com"
}
返回：
{
	"username": "Jonas Vingegaard",
	"email": "test@test.com",
	"full_name": null
}

发送：
{
    "username": "Jonas Vingegaard",
    "password": "123456"
}
返回：
{
	"detail": [
		{
			"type": "missing",
			"loc": [
				"body",
				"email"
			],
			"msg": "Field required",
			"input": {
				"username": "Jonas Vingegaard",
				"password": "123456"
			}
		}
	]
}

8，使用数据库

任何SQLAlchemy支持的数据库，如：

  • PostgreSQL
  • MySQL
  • SQLite
  • Oracle
  • Microsoft SQL Server，等等其它数据库

首先你需要安装SQLAlchemy:

pip install sqlalchemy

创建目录：

└── sql_app
    ├── __init__.py
    ├── crud.py
    ├── database.py
    ├── main.py
    ├── models.py
    └── schemas.py

定义数据库：

# sql_app/database.py

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

# 为 SQLAlchemy 定义数据库 URL地址
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

# 创建 SQLAlchemy 引擎
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)

# 创建一个SessionLocal类
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 创建一个Base类
Base = declarative_base()

创建数据库模型:

# sql_app/models.py

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String

from sql_app.database import Base


class User(Base):
    __tablename__ = "users"
    __table_args__ = {'extend_existing': True}

    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

创建 Pydantic 模型:

# sql_app/schemas.py

from pydantic import BaseModel


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool

    class Config:
        orm_mode = True

主程序——controller：

# main.py

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from sql_app import crud, models, schemas
from sql_app.database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


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


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user



if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

添加 crud 操作实现：

# crud.py

from sqlalchemy.orm import Session

from sql_app import models, schemas


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

去启动程序，访问 API 页面：
在这里插入图片描述
POST 创建一些数据，然后就能 GET 它们了。

9，文件的上传与下载

简单的文件上传：

from urllib.parse import quote

from fastapi import FastAPI, UploadFile
from fastapi.responses import StreamingResponse

app = FastAPI()


@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):
    # 分块将大文件保存到磁盘
    file_object = file.file
    file_name = file.filename
    # 判断文件夹是否存在，不存在则创建
    import os
    if not os.path.exists("files"):
        os.mkdir("files")
    # 将文件写入到文件夹
    with open(f"files/{file_name}", "wb") as f:
        while True:
            # 读取文件
            chunk = file_object.read(10000)
            if not chunk:
                break
            f.write(chunk)

    return {"filename": file.filename}


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

大文件下载：


def file_iterator(file_path, chunk_size=8192):
    with open(file_path, mode="rb") as file_like:
        while True:
            chunk = file_like.read(chunk_size)
            if not chunk:
                break
            yield chunk


# 流式下载文件
@app.get("/downloadfile/")
async def download_file(file_name: str):
    file_path = f"files/{file_name}"
    # 判断文件是否存在
    import os
    if not os.path.exists(file_path):
        return {"msg": "file not found"}
    else:
        headers = {
            'Content-Disposition': f'attachment; filename="{quote(file_name)}"'
        }
        return StreamingResponse(
            file_iterator(file_path),
            media_type="application/octet-stream",
            headers=headers
        )

10，部署 fastapi 项目

先收集项目依赖：

pip freeze > requirements.txt

然后将项目发送到服务器。
在服务器安装 ASGI 服务器：

pip install "uvicorn[standard]"

然后运行服务器程序：

uvicorn main:app --host 0.0.0.0 --port 80

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：/a/734117.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

Excel 如何复制单元格而不换行

Excel 如何复制单元格而不换行

1. 打开excle, sheet1右键单击>查看代码>插入>模块 输入代码 Sub CopyText() Updated by NirmalDim xAutoWrapper As ObjectSet xAutoWrapper New DataObject or GetObject("New:{1C3B4210-F441-11CE-B9EA-00AA006B1A69}")xAutoWrapper.SetText ActiveC…
阅读更多...
数据库精选题（一）（关系数据库设计）

数据库精选题（一）（关系数据库设计）

🌈 个人主页：十二月的猫-CSDN博客 🔥 系列专栏： 🏀数据库 💪🏻 十二月的寒冬阻挡不了春天的脚步，十二点的黑夜遮蔽不住黎明的曙光 目录 前言 练习题 题型一：判断关系…
阅读更多...
【CV炼丹师勇闯力扣训练营 Day8】

【CV炼丹师勇闯力扣训练营 Day8】

CV炼丹师勇闯力扣训练营 代码随想录算法训练营第8天 ● 344.反转字符串 ● 541. 反转字符串II ● 卡码网：54.替换数字 一、344 反转字符串 编写一个函数，其作用是将输入的字符串反转过来。输入字符串以字符数组 s 的形式给出。 不要给另外的数组分配额…
阅读更多...
Redis实战—Redis分布式锁

Redis实战—Redis分布式锁

本博客为个人学习笔记，学习网站与详细见：黑马程序员Redis入门到实战 P56 - P63 目录 分布式锁介绍 基于Redis的分布式锁 Redis锁代码实现 修改业务代码 分布式锁误删问题 分布式锁原子性问题 Lua脚本 编写脚本 代码优化 总结 分布式锁介绍…
阅读更多...
【技巧】Leetcode 201. 数字范围按位与【中等】

【技巧】Leetcode 201. 数字范围按位与【中等】

数字范围按位与 给你两个整数 left 和 right ，表示区间 [left, right] ，返回此区间内所有数字 按位与 的结果（包含 left 、right 端点）。 示例 1： 输入：left 5, right 7 输出：4 解题思路 …
阅读更多...
vscode禅模式怎么退出

vscode禅模式怎么退出

1、如何进入禅模式：查看--外观--禅模式 2、退出禅模式 按二次ESC，就可以退出。
阅读更多...
公共 IP 地址和私有 IP 地址的区别总结

公共 IP 地址和私有 IP 地址的区别总结

什么是IP地址？ IP 地址，即互联网协议地址（Internet Protocol Address），是网络设备在网络中进行通信的标识。IP 地址可以看作是设备在网络中的“地址”，有助于数据包在网络中找到正确的接收端。IP 地址主要…
阅读更多...
计算机系统基础实训七-MallocLab实验

计算机系统基础实训七-MallocLab实验

实验目的与要求 1、让学生理解动态内存分配的工作原理； 2、让学生应用指针、系统级编程的相关知识； 3、让学生应用各种动态内存分配器的实现方法； 实验原理与内容 （1）动态内存分配器基本原理 动态内存分配器维护…
阅读更多...
外包IT运维解决方案

外包IT运维解决方案

随着企业信息化进程的不断深入，IT系统的复杂性和重要性日益增加。高效的IT运维服务对于保证业务连续性、提升企业竞争力至关重要。外包IT运维解决方案通过专业的服务和技术支持，帮助企业降低运维成本、提高运维效率和服务质量。 本文结合《外包IT运维解…
阅读更多...
咖啡事故，上海Manner咖啡店，1天两起店员和顾客发生冲突

咖啡事故，上海Manner咖啡店，1天两起店员和顾客发生冲突

上海咖啡店Manner，一天的时间竟然发生两起店员和员工发生肢体冲突： 事情详情： Manner威海路716店事件: 店员泼顾客咖啡粉，随后被辞退品牌方回应媒体，表示将严肃处理Manner梅花路门店事件:顾客因等待时间长抱怨&…
阅读更多...
Aquila-Med LLM：开创性的全流程开源医疗语言模型

Aquila-Med LLM：开创性的全流程开源医疗语言模型

​论文链接：https://arxiv.org/pdf/2406.12182 开源链接：https://huggingface.co/BAAI/AquilaMed-RL http://open.flopsera.com/flopsera-open/details/AquilaMed_SFT http://open.flopsera.com/flopsera-open/details/AquilaMed_DPO 近年来&#xf…
阅读更多...
Magento1与Magento2的区别

Magento1与Magento2的区别

本人接触magento有些年头了。。。 2012年开始用magento 1.7。2016年开始用magento2.0。 截止到目前。M1最新版本是1.9.3.3。 M2最新版本是2.2.2。 想当年第一次接触magento的时候，是跟同事一起，网上下载的Alan Storm的深入理解magento系统，…
阅读更多...
链表中环的入口节点

链表中环的入口节点

链表中环的入口节点 描述 链表中环的入口节点 给一个长度为n链表&#xff0c;若其中包含环&#xff0c;请找出该链表的环的入口结点&#xff0c;否则&#xff0c;返回null。 数据范围&#xff1a; n≤10000&#xff0c; 1<结点值<10000 要求&#xff1a;空间复杂度 O(1)…
阅读更多...
windows下mysql修改 my.ini的datadir后 `Access denied`

windows下mysql修改 my.ini的datadir后 `Access denied`

1. 背景 window安装mysql数据库时，不能指定数据文件存放位置（默认安装路径 "C:/ProgramData"）。 只能通过修改mysql.ini来更改数据文件存放目录。 2. 问题： 修改mysql.ini后，mysql 出现 "Access denied for user ‘root‘@‘localhost‘ (using passwor…
阅读更多...
webpack安装sass

webpack安装sass

package.json文件 {"devDependencies": {"sass-loader": "^7.2.0","sass": "^1.22.10","fibers": "^4.0.1"} }这个不用webpack.config.js module.exports {module: {rules: [{test: /\.s[ac]ss$/i,u…
阅读更多...
算法设计与分析：分治法求最近点对问题

算法设计与分析：分治法求最近点对问题

目录 一、实验目的 二、实验内容 三、算法思想 四、实验步骤 1、蛮力法 2、分治法 2.1 先用快速排序SortX(A,1,n)将所有点按x坐标升序排序 2.2 点数n<3时直接计算&#xff0c;时间复杂度为O(1) 2.3 点数n>3时 五、实验结果和分析 一、实验目的 1. 掌握分治法思…
阅读更多...
Ilya出走记：SSI的超级安全革命

Ilya出走记：SSI的超级安全革命

图片&#xff5c;OpenAI官网 ©自象限原创 作者丨罗辑、程心 和OpenAI分道扬镳以后&#xff0c;Ilya“神秘而伟大”的事业终于揭开了面纱。 6月20日&#xff0c;前OpenAI核心创始人 Ilya Stuskever&#xff0c;在官宣离职一个月后&#xff0c;Ilya在社交媒体平台公开了…
阅读更多...
SambaLingo——教会大模型新语言

SambaLingo——教会大模型新语言

在当今数字化时代&#xff0c;语言不仅是沟通的桥梁&#xff0c;也是信息和知识传递的核心。尽管大模型&#xff08;LLMs&#xff09;在处理英语等主流语言方面取得了显著进展&#xff0c;但它们在理解和生成其他语言内容方面的能力却参差不齐。这种不平衡限制了技术在全球范围…
阅读更多...
Charles抓取安卓应用https包演示

Charles抓取安卓应用https包演示

一、准备软件 夜神安卓模拟器 (yeshen.com) Charles (charlesproxy.com) 二、配置抓包 2.1 Charles安装PC根证书 记住这里的ip端口 三、安卓模拟器配置 3.1 配置安卓客户端网络代理 填写上文的ip端口&#xff0c;保存 3.2 安装根证书 3.2.1 导出根证书 linux主机执行 op…
阅读更多...
Springboot项目ES报异常query_shard_exception

Springboot项目ES报异常query_shard_exception

详细异常信息如下&#xff1a; {"error": {"root_cause": [{"type": "query_shard_exception","reason": "failed to create query: {\n \"bool\" : {\n \"filter\" : [\n {\n \…
阅读更多...
最新文章

Copyright @ 2022~2023