fastapi-pydantic
综述
首先我们应该思考一个web框架有哪些大部分组成,网页web的功能就是把持久化的数据提供为http服务
之后随着深入理解,我会不断更新这个东西,有可能会独立出一篇文章来
一门语言里面有什么?一个web框架里面应该有什么?主要功能点列表
- 路由(Routing)
- 请求-响应处理
- 中间件(Middleware)
- 模板引擎(Template Engine)
- 静态文件处理
- 会话管理(Session Management)
- 安全性
- 数据验证
- 数据库支持
- 测试支持
- 国际化和本地化
- 缓存支持
- 任务调度
- 日志记录
Gunicorn与uWSGI之我见 · 跬步
Gunicorn的使用手册看这篇就够了【用过都说好】-云社区-华为云
WSGI和ASGI的异同_studyeboy的博客-CSDN博客_asgi和wsgi的区别
快如闪电的异步服务器Uvicorn - 知乎
WSGI 理解 · 跬步
WSGI与Tornado想到的 · 跬步
Gunicorn的使用手册看这篇就够了【用过都说好】-云社区-华为云
Cookie Session Token JWT的区别和实现原理(2) - JustJavaIt - 博客园
普通登录、微服务登录、单点登录的区别_哔哩哔哩_bilibili session
- Get方法长度限制
Http Get方法提交的数据大小长度并没有限制,HTTP协议规范没有对URL长度进行限制。这个限制是特定的浏览器及服务器对它的限制。
浏览器 | 字符长度限制 |
---|---|
Microsoft Internet Explorer (Browser) | 2083 如果超过,提交按钮没有任何反应 |
Firefox (Browser) | 65,536 |
Safari (Browser) | 80,000 |
Opera (Browser) | 190,000 |
Google (chrome) | 8182 |
Apache (Server) | 8,192 |
Microsoft Internet Information Server(IIS) | 16,384 |
由此,URL最好不要超过IE的最大长度限制(2083个字符),当然,如果URL不直接提供给用户,而是提供给程序调用,这时的长度就只受Web服务器影响了
POST请求理论上也是不限制大小的,真正对其大小进行限制的是服务器的处理程序能力。Tomcat服务器对POST大小限制为2M
{"id":string4-20, "string_list":[string(4-20)100]} * 2000 条 fastapi的POST数据就buffer溢出报错
API和SPI
- API设计指南 通俗的讲:API 和 SPI 都是相对的概念,他们的差别只在语义上,API 直接被应用开发人员使用,SPI 被框架扩展人员使用
API (Application Programming Interface) - 大多数情况下,都是实现方来制定接口并完成对接口的不同实现,调用方仅仅依赖却无权选择不同实现。 SPI (Service Provider Interface)
- 而如果是调用方来制定接口,实现方来针对接口来实现不同的实现。调用方来选择自己需要的实现方。 需要注意的事项 SPI 和 API 也不一定是接口,我这里都是指狭义的具体的接口。 GitHub - ibbd-dev/fastapi-start: FastAPI脚手架:用于系统后端接口项目的
fastapi中async
是否使用async[1]
使用async和不使用async
from fastapi import FastAPI
import time
import asyncio
import os
app = FastAPI()
@app.get("/async_slowest")
async def async_slowest():
time.sleep(1)
return {"message": "async mode but use sync sleep"}
@app.get("/async_sleep_in_thread")
async def async_sleep_in_thread():
loop = asyncio.get_event_loop()
await loop.run_in_executor(None, time.sleep, 1)
return {"message": "sleep run in thread pool"}
@app.get("/async_sleep")
async def async_sleep():
await asyncio.sleep(1)
return {"message": "async mode sleep"}
@app.get("/sync")
def sync_sleep():
time.sleep(1)
return {"message": "sync, but run in thread pool"}
分别执行100次
模式 | 原理 | 时间 |
---|---|---|
异步函数同步方法 | 会被扔到eventloop同步执行 | 100s |
异步函数线程执行run_in_executor | 扔到线程池里执行开100个线程 | 1s |
异步函数异步方法 | 就是纯纯异步 | 1s |
同步函数同步方法 | 自动线程池执行,线程池的默认核心数*5 | 3s |
官方文档的细节[2] #todo 看完 FastAPI会对路径操作函数(path operation function)和依赖(dependencies)进行特殊处理。这个特殊处理是:如果你把函数定义为def而不是async def,那么FastAPI会把它放到单独的线程池中,异步执行,这就是FastAPI精彩的地方。就像官方所说,如果你不清楚你函数里面的调用是不是异步(能不能用await),那么就把它定义为普通函数,FastAPI会采用多线程的方式处理。乱用async,在async里面有同步调用,则会变成串行,Fast秒变Slow。
而对于其他函数,FastAPI则不会管,def就是同步调用,立马返回结果。
现在回过头来看前面的那句话:但是无论你是否使用async,FastAPI都将异步工作,以达到"Fast"的运行速度。
pydantic
pydantic比dataclass以及其他方法都慢得多,无论是在创建还是在修改的时候,pydantic因为在runtime会进行解析,创建需要5400ns,但例如tuple只需要30ns,慢了一百八十倍。dataclass需要250ns,慢了21倍。
但是好用啊!
序列化
pydantic的基本使用就不说了,继承Base剩下的就是定义字段名字。
字段的名字就是序列化字典和json的名字 ,Field中可以使用alias参数
在使用alias之后,parse的时候读取的就是alias字段。
序列化的时候json和dict,指定by_alias键就是alias,否则就是定义的字段名字。
Click to see more
from pydantic import BaseModel, Field
class R(BaseModel):
code: int = Field(alias="Code")
message: str = Field(alias="Mess")
data: str = Field(alias="Data")
commited: int = 1
r = R.parse_obj({"Code": 1, "Mess": "abc", "Data": "abc", "commited": 2})
y=r.dict(by_alias=True)
x = r.json()
print(x,y)
运行结果
{"code": 1, "message": "abc", "data": "abc", "commited": 2} {'Code': 1, 'Mess': 'abc', 'Data': 'abc', 'commited': 2}
常用的还有自定义json序列化器,可以按照类型序列化,也可以拦截自定义处理extra = Extra.allow
允许创建完model后还可以添加属性
使用json序列化器的时候,使用encoders只会去序列化非内置字段,builtin的类型会直接去序列化。
对于所有的类型都序列化的话,有两种方法
- 使用json_dumps 参数,会把所有的字段传进去,它会比指定的encoder早返回的是json编码的解码?,encoder会失效,
- @validator这种方法还没实验
Click to see more
# -*-coding:utf-8-*-
# SettingCode here
__author__ = "a_little_rubbish"
__date__ = "2023/1/9 21:49"
# import your model here
from datetime import datetime
import orjson
from pydantic import BaseModel, Field, Extra
from utils.schema import convert_date
def convert_datetime(dt) -> str: # 拦截器
print(dt.strftime('%Y-%m-%d %H:%M:%S'))
return dt.strftime('%Y-%m-%d %H:%M:%S')
def ascii_to_uft8(v, *, default):
for key, value in v.items():
if isinstance(value, str):
print("str hook")
v[key] = value.encode('utf-8')
return orjson.dumps(v, default=default).decode() # 注意返回什么
class TemplateBaseModel(BaseModel):
class Config:
extra = Extra.allow # 允许创建完model后还可以添加属性
arbitrary_types_allowed = True # 允许自定义类型
json_dumps = ascii_to_uft8 # 指定json序列化器,可以做钩子
orm_mode = True
json_encoders = { # 对特定类型的json编码器
# custom output conversion for datetime
datetime : convert_datetime
}
class R(TemplateBaseModel):
code: int = Field(alias="Code")
message: str = Field(alias="Mess")
data: str = Field(alias="Data")
commited: int = 1
r = R.parse_obj({"Code": 1, "Mess": "abc", "Data": "abc", "commited": 2})
y=r.dict(by_alias=True) # 使用alias
x = r.json()
print(x,y)
运行结果
str hook
str hook
{"code":1,"message":"abc","data":"abc","commited":2} {'Code': 1, 'Mess': 'abc', 'Data': 'abc', 'commited': 2}
默认生成uuid不生效
问题:有一个字段不是前端传过来的,需要赋一个默认值uuid,但是发现穿进来的数据每一批的uuid都一样,不用批次的uuid有时候一样有时候不一样
原因:验证器生效在默认值之前,也就是说pydantic先验证再给默认值,而且每个pydantic对象只初始化一次,这样就导致uuid字段被固定了下来,同一批次的数据都被验证器赋了同一个uuid。不用批次的数据可能会被多进程(线程?)处理,如果两次都是同一个线程,那就uuid相同不同线程uuid就不同
#todo Fastapi系统的整理一下 高性能 FastAPI 框架入门精讲-慕课网-Python Web 框架
Fastapi框架实战系列-纯个人使用之脚手架搭建篇 - 墨天轮
使用jwt的时候生成jwt的16进制字符串密钥
import secrets
print(secrets.token_hex(32)) # 生成一个安全的随机密钥
pgsql->redis->rabbit->mongo->es->minio fastapi->python->sqlalchemy->celery->pandas和numpy