php源码网站修改,织梦做的网站如何上线,有可以花钱让人做问券的网站吗,旅行社erp系统FastAPI 是一个用于构建 API 的现代、快速#xff08;高性能#xff09;的 web 框架#xff0c;使用 Python 并基于标准的 Python 类型提示。
关键特性:
快速#xff1a;可与 NodeJS 和 Go 并肩的极高性能#xff08;归功于 Starlette 和 Pydantic#xff09;。最快的 …FastAPI 是一个用于构建 API 的现代、快速高性能的 web 框架使用 Python 并基于标准的 Python 类型提示。
关键特性:
快速可与 NodeJS 和 Go 并肩的极高性能归功于 Starlette 和 Pydantic。最快的 Python web 框架之一。高效编码提高功能开发速度约 200 至 300。更少 bug减少约 40 的人为开发者导致错误。智能极佳的编辑器支持。处处皆可自动补全减少调试时间。简单设计的易于使用和学习阅读文档的时间更短。简短使代码重复最小化。通过不同的参数声明实现丰富功能。bug 更少。健壮生产可用级别的代码。还有自动生成的交互式文档。标准化基于并完全兼容API 的相关开放标准OpenAPI (以前被称为 Swagger) 和 JSON Schema。
简单来体验一下 环境
Python 3.11Windows 10fastapi 0.68.0
1安装 fastapi
pip install fastapi2安装一个 ASGI 服务器
pip install uvicorn[standard]生产环境可以使用 Uvicorn 或者 Hypercorn。 ASGI(Asynchronous Server Gateway Interface)是 Python 异步服务器网关接口的缩写。 它定义了异步编程风格的 Python Web 服务器与 Web 服务器和应用程序之间的标准接口。 ASGI 旨在解决传统 Python Web 服务器(如WSGI)在高并发和长连接场景下的不足使用异步编程模型来提高服务器的并发处理能力。 ASGI服务器具有以下几个关键特点: 异步和事件驱动ASGI 服务器采用异步非阻塞的编程模型可以高效处理大量并发连接避免进程/线程阻塞。它使用事件循环机制在有事件到来时处理否则释放资源做其他事情。支持长连接和WebSocketASGI 服务器本身支持长连接和 WebSocket 协议可用于构建实时通信应用如在线聊天室、协作平台等。 标准统一接口ASGI 定义了统一的服务器端接口规范使Web服务器、中间件和应用程序之间松散耦合易于拓展和维护。 高性能相比传统 WSGIASGI 编程模型使用异步协程可以大幅降低上下文切换和内存开销从而提高并发性能。 目前支持 ASGI 的服务器有 Uvicorn、Daphne、Hypercorn 等应用框架包括 Django Channels、FastAPI 等。ASGI 正在逐渐取代 WSGI 成为 Python Web 应用开发的新标准特别适合于需要高并发、实时双向通信的应用场景。 3创建一个 main.py 文件
from typing import Unionfrom 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 CTRLC to quit)
INFO: Started reloader process [14776] using WatchFiles
INFO: Started server process [24944]
INFO: Waiting for application startup.
INFO: Application startup complete.mainmain.py 文件一个 Python “模块”。app在 main.py 文件中通过 app FastAPI() 创建的对象。–reload让服务器在更新代码后重新启动。仅在开发时使用该选项。
5浏览器访问路由
访问 http://127.0.0.1:8000/ 可以在页面看到路由处理函数返回的数据 访问 http://127.0.0.1:8000/items/5?qsomequery 可以在页面看到路由处理函数返回的数据
6体验类型声明的优势
浏览器访问 http://127.0.0.1:8000/docs 可以到达 API 界面 但是国内可能打不开。 fastapi访问/docs和/redoc接口文档显示空白或无法加载 FastAPI 自动文档Swagger UI 打不开。显示空白 打开这个页面我们可以看到非常详细的 API 描述
item_idinteger类型是一个路径参数而且是必需的。qstring 类型是一个查询参数而且不是必需的。
这是怎么做到的回到代码
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 即可声明可选的查询参数
最重要的特征就是我们为参数添加了类型声明这样做有非常多好处
由 FastAPI 通过类型声明自动将解析请求中的字符串转为声明的数据类型 2更方便地利用编辑器的自动提示
7借助 Pydantic 完成数据验证
多数情况我们会用 POST最常用、PUT 操作来发送数据具体就是将数据放在这些请求的请求体中。 既然路径参数和查询参数都可以通过定义数据类型来完成默认的类型转化和校验那么作为传述数据主要载体的请求体自然也支持数据类型的定义带来的好处响应体也是如此
请求体和响应体的数据定义需要使用 pandatic这是一个流行的数据验证工具。使用 Pydantic 模型声明请求体能充分利用它的功能和优点
以 JSON 形式读取请求体。在必要时把请求体转换为对应的类型校验数据数据无效时返回错误信息并指出错误数据的确切位置和内容把接收的数据赋值给参数 item把函数中请求体参数的类型声明为 Item还能获得代码补全等编辑器支持为模型生成 JSON Schema在项目中所需的位置使用用于 API 文档 UI
from fastapi import FastAPI
from pydantic import BaseModelclass Item(BaseModel):name: strdescription: str | None Noneprice: floattax: float | None Noneapp 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 Anyfrom fastapi import FastAPI
from pydantic import BaseModel, EmailStrapp FastAPI()class UserIn(BaseModel):username: strpassword: stremail: EmailStrfull_name: str | None Noneclass UserOut(BaseModel):username: stremail: EmailStrfull_name: str | None Noneapp.post(/user/, response_modelUserOut)
async def create_user(user: UserIn) - Any:return userif __name__ __main__:import uvicornuvicorn.run(app)即便我们的路径操作函数将会返回包含密码的相同输入用户但我们已经将 response_model 声明为了不包含密码的 UserOut 模型FastAPI 将会负责过滤掉未在输出模型中声明的所有数据。
发送
{username: Jonas Vingegaard,password: 123456,email: testtest.com,full_name: Jonas Vingegaard Rasmussen
}
返回
{username: Jonas Vingegaard,email: testtest.com,full_name: Jonas Vingegaard Rasmussen
}发送
{username: Jonas Vingegaard,password: 123456,email: testtest.com
}
返回
{username: Jonas Vingegaard,email: testtest.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支持的数据库如
PostgreSQLMySQLSQLiteOracleMicrosoft SQL Server等等其它数据库
首先你需要安装SQLAlchemy:
pip install sqlalchemy创建目录
└── sql_app├── __init__.py├── crud.py├── database.py├── main.py├── models.py└── schemas.py定义数据库
# sql_app/database.pyfrom 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:passwordpostgresserver/db# 创建 SQLAlchemy 引擎
engine create_engine(SQLALCHEMY_DATABASE_URL, connect_args{check_same_thread: False}
)# 创建一个SessionLocal类
SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine)# 创建一个Base类
Base declarative_base()创建数据库模型:
# sql_app/models.pyfrom sqlalchemy import Boolean, Column, ForeignKey, Integer, Stringfrom sql_app.database import Baseclass User(Base):__tablename__ users__table_args__ {extend_existing: True}id Column(Integer, primary_keyTrue)email Column(String, uniqueTrue)hashed_password Column(String)is_active Column(Boolean, defaultTrue)
创建 Pydantic 模型:
# sql_app/schemas.pyfrom pydantic import BaseModelclass UserBase(BaseModel):email: strclass UserCreate(UserBase):password: strclass User(UserBase):id: intis_active: boolclass Config:orm_mode True主程序——controller
# main.pyfrom fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from sql_app import crud, models, schemas
from sql_app.database import SessionLocal, enginemodels.Base.metadata.create_all(bindengine)app FastAPI()# Dependency
def get_db():db SessionLocal()try:yield dbfinally:db.close()app.post(/users/, response_modelschemas.User)
def create_user(user: schemas.UserCreate, db: Session Depends(get_db)):db_user crud.get_user_by_email(db, emailuser.email)if db_user:raise HTTPException(status_code400, detailEmail already registered)return crud.create_user(dbdb, useruser)app.get(/users/, response_modellist[schemas.User])
def read_users(skip: int 0, limit: int 100, db: Session Depends(get_db)):users crud.get_users(db, skipskip, limitlimit)return usersapp.get(/users/{user_id}, response_modelschemas.User)
def read_user(user_id: int, db: Session Depends(get_db)):db_user crud.get_user(db, user_iduser_id)if db_user is None:raise HTTPException(status_code404, detailUser not found)return db_userif __name__ __main__:import uvicornuvicorn.run(app)
添加 crud 操作实现
# crud.pyfrom sqlalchemy.orm import Sessionfrom sql_app import models, schemasdef 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 notreallyhasheddb_user models.User(emailuser.email, hashed_passwordfake_hashed_password)db.add(db_user)db.commit()db.refresh(db_user)return db_user
去启动程序访问 API 页面 POST 创建一些数据然后就能 GET 它们了。
9文件的上传与下载
简单的文件上传
from urllib.parse import quotefrom fastapi import FastAPI, UploadFile
from fastapi.responses import StreamingResponseapp FastAPI()app.post(/uploadfile/)
async def create_upload_file(file: UploadFile):# 分块将大文件保存到磁盘file_object file.filefile_name file.filename# 判断文件夹是否存在不存在则创建import osif not os.path.exists(files):os.mkdir(files)# 将文件写入到文件夹with open(ffiles/{file_name}, wb) as f:while True:# 读取文件chunk file_object.read(10000)if not chunk:breakf.write(chunk)return {filename: file.filename}if __name__ __main__:import uvicornuvicorn.run(app)
大文件下载 def file_iterator(file_path, chunk_size8192):with open(file_path, moderb) as file_like:while True:chunk file_like.read(chunk_size)if not chunk:breakyield chunk# 流式下载文件
app.get(/downloadfile/)
async def download_file(file_name: str):file_path ffiles/{file_name}# 判断文件是否存在import osif not os.path.exists(file_path):return {msg: file not found}else:headers {Content-Disposition: fattachment; filename{quote(file_name)}}return StreamingResponse(file_iterator(file_path),media_typeapplication/octet-stream,headersheaders)10部署 fastapi 项目
先收集项目依赖
pip freeze requirements.txt然后将项目发送到服务器。 在服务器安装 ASGI 服务器
pip install uvicorn[standard]然后运行服务器程序
uvicorn main:app --host 0.0.0.0 --port 80
