1、框架对比

https://blog.csdn.net/qq_39241986/article/details/115423474
https://www.cnblogs.com/zhuminghui/p/14741536.html
　从软件包，社区，性能，灵活性，职位空缺和培训来进行比较。

软件包丰富程度——Django 具有使代码可重用的大多数软件包，是一个完整的 Web 开发框架，而 Flask 和 FastAPI 是用于构建网站的简约框架，很多功能比如用户系统，后台管理要自己实现。
社区活跃程度——Django 社区是最活跃的社区，这是因为它使用广泛，很多大厂使用，另一方面，Flask 的社区也很繁荣，仅次于 Django。FastAPI 的社区目前还比较小，因为它相对较新。
性能。在性能方面——FastAPI 是领跑者，因为它是面向速度的，其次是 Flask，最后是 Django。
灵活性——灵活性是开发人员非常重视的东西，并且 Flask 比 Django 更灵活。另一方面，FastAPI 在代码方面是灵活的，并且不限制代码布局。因此，我们可以说 Flask 在这三者中是最灵活的。
职位空缺——毫无疑问，Python 网络生态系统中有 Django 要求的职位空缺更多，其次是 Flask，最后是 FastAPI，其数量要少得多，因此，如果你打算快速找到工作，那么 Django 应该是首选。
学习成本——FastAPI < Flask < Django。Django 虽然学习起来比较费劲，但是有完善的官方文档和很多在线资料和资源。Flask 既简单又直接，也有丰富的在线资料和资源。而 FastAPI 学习起来更简单直接，不过资源相对较少，因为还需要时间。

2、asgi VS wsgi 区别

参考：https://zhuanlan.zhihu.com/p/659198893
ASGI（Asynchronous Server Gateway Interface）和 WSGI（Web Server Gateway Interface）都是 Python Web 应用程序与 Web 服务器之间的接口标准，但它们有一些重要的区别：

1. 同步 vs. 异步：

• WSGI 是一个同步接口，这意味着它是基于同步 I/O 的。在 WSGI 应用程序中，每个请求都是按顺序处理的，一个请求需要等待另一个请求完成后才能继续。这对于低并发的应用程序是足够的，但在高并发场景下可能会限制性能。

• ASGI 是一个异步接口，支持异步 I/O。ASGI 应用程序可以处理多个请求，而无需等待一个请求完成。这使得 ASGI 更适合于高并发和实时性要求高的应用程序，例如聊天应用、实时通知、在线游戏等。

2. WebSocket 和长轮询支持：

• ASGI 支持 WebSocket 连接和长轮询（Long Polling），使得实时通信更容易实现。

• WSGI 不直接支持 WebSocket，虽然可以通过扩展和库来实现，但相对复杂。

3. 性能差异：

• 由于异步处理的能力，ASGI 可以更好地处理高并发情况，因此在某些场景下性能可能更好。

• WSGI 在处理低并发和传统 Web 应用程序时表现良好，但在高并发场景下可能会有限制。

4. 中间件和组件：

• ASGI 和 WSGI 都支持中间件和组件，但由于异步性质的不同，ASGI 中的中间件可能需要以异步方式编写。

• WSGI 中的中间件通常是同步的。

5. 部署选项：

• WSGI 应用程序可以在广泛的 Web 服务器上运行，因为许多 Web 服务器都支持 WSGI 接口。

• ASGI 应用程序通常需要专门支持 ASGI 标准的服务器，例如 Daphne、Uvicorn、Hypercorn 等。

总的来说，ASGI 更适合处理高并发、实时通信和异步任务处理等需求，而 WSGI 更适合传统的 Web 应用程序。在选择 ASGI 还是 WSGI 时，需要根据你的应用程序的性质和性能需求来做出决策。如果你的应用程序需要实时通信、WebSocket 支持或需要处理大量并发请求，那么考虑使用 ASGI 可能是一个明智的选择。否则，WSGI 可能足够满足你的需求。

3、知识点

中文文档：https://fastapi.tiangolo.com/zh/#_10
GitHub上案例：https://github.com/mjhea0/awesome-fastapi?tab=readme-ov-file#utils
pydantic官方文档:
https://hellowac.github.io/pydantic-zh-cn/v1.10.7-zh-cn/
https://docs.pydantic.dev/latest/
1、pydantic models 生成工具
https://github.com/koxudaxi/fastapi-code-generator
https://github.com/koxudaxi/datamodel-code-generator

4、启动方式

1、FastAPI CLI启动
https://fastapi.tiangolo.com/fastapi-cli/
FastAPI CLI是一个命令行程序，您可以使用它来提供您的 FastAPI 应用程序、管理您的 FastAPI 项目等等。
当你安装 FastAPI（例如使用pip install fastapi）时，它包含一个名为的包fastapi-cli，此包提供fastapi终端中的命令。

• fastapi dev
  当您运行时fastapi dev，它将以开发模式运行。默认情况下，它会启用自动重新加载，因此当您更改代码时，它会自动重新加载服务器。这会占用大量资源，并且稳定性可能不如没有它时那么稳定，您应该只在开发时使用它。
• fastapi run
  当你运行时fastapi run，它将默认以生产模式运行。默认情况下它将禁用自动重新加载。它将监听 IP 地址0.0.0.0，也就是所有可用的 IP 地址，这样任何可以与机器通信的人都可以公开访问它。这是您在生产中通常运行它的方式，例如在容器中。

2、uvicorn 启动
在内部，FastAPI CLI使用uvicorn ，一款高性能、可用​​于生产的 ASGI 服务器。
https://www.uvicorn.org/

• 从命令行运行¶
  uvicorn通常您将从命令行运行。

$ uvicorn main:app --reload --port 5000

ASGI 申请应以 的形式指path.to.module:instance.path。
在本地运行时，使用–reload开启自动重新加载。
–reload和参数–workers是互相排斥的。
Uvicorn 包含一个–workers允许您运行多个工作进程的选项。

$ uvicorn main:app --workers 4

与 gunicorn 不同，uvicorn 不使用 pre-fork，而是使用spawn，这使得 uvicorn 的多进程管理器在 Windows 上仍然可以很好地运行。一般用gunicorn 。

• 以编程方式运行
  要直接从 Python 程序中运行，您应该使用uvicorn.run(app, **config)。例如：

import uvicorn
class App:...
app = App()if __name__ == "__main__":uvicorn.run("main:app", host="127.0.0.1", port=5000, log_level="info")

from typing import Union
from fastapi import FastAPI
import uvicorn
app = FastAPI()@app.get("/")
def read_root():return {"Hello": "World"}@app.get("/items/{item_id}")
def read_item(item_id: int, q: Union[str, None] = None):return {"item_id": item_id, "q": q}if __name__ == "__main__":config = uvicorn.Config("main:app", port=5000, log_level="info")server = uvicorn.Server(config)server.run()

5、Pydantic 模型

字段类型参考：typing和pydantic的
https://docs.pydantic.dev/1.10/usage/types/#literal-type
https://docs.pydantic.dev/2.0/usage/models/#nested-models
FastAPI 站在以下巨人的肩膀之上：
Starlette 负责 web 部分。
Pydantic 负责数据部分。

pydantic库是一种常用的用于数据接口schema定义与检查的库。Pydantic 在运行时强制执行类型提示，并在数据无效时提供用户友好的错误信息。

Pydantic 模型:
Pydantic 是一个用来用来执行数据校验的 Python 库。
你可以将数据的"结构"声明为具有属性的类。每个属性都拥有类型。接着你用一些值来创建这个类的实例，这些值会被校验，并被转换为适当的类型（在需要的情况下），返回一个包含所有数据的对象。

python数据类型：

https://blog.csdn.net/weixin_49520696/article/details/134172661
python基本数据类型：int、float、bool、bytes
Python 内置容器： dict、list、set 和 tuple
Python 泛型：（typing模块）
Python 的泛型用于指示某些数据结构或函数可以接受多种数据类型，例如使得一个函数的参数能够接收多种类型的数据或者使得一个类能够接收并存储不同种类的数据。泛型在 Python 中通常通过模块 typing 实现。泛型的使用有助于提高代码的可读性和健壮性，尤其是在大型项目和团队协作中。
Python 中的泛型是使用 typing 模块中的 TypeVar 和 Generic 进行实现的。TypeVar 用于定义泛型类型变量，而 Generic 用于定义泛型类或函数。
typing 模块中的泛型支持包括一系列的泛型类型和类型变量，例如 List、Dict、Tuple 等。开发者可以使用这些泛型类型来声明具有泛型参数的数据结构或函数签名。此外，Python 3.9 引入了更多强大的泛型支持，包括 Literal、TypedDict 等新的泛型类型。
需要注意的是，Python 中的泛型类型提示仅用于静态类型检查和文档说明，并不影响运行时的行为。Python 解释器不会强制执行类型提示，因此在运行时仍然可以传入任何类型的参数。

from typing import TypeVar, GenericT = TypeVar('T')def first(items: List[T]) -> T:return items[0]print(first([1, 2, 3]))  # 1
print(first(["apple", "banana", "cherry"]))  # apple

from typing import TypeVar, GenericT = TypeVar('T')class Stack(Generic[T]):def __init__(self) -> None:self.items: List[T] = []def push(self, item: T) -> None:self.items.append(item)def pop(self) -> T:return self.items.pop()stack = Stack[int]()
stack.push(1)
stack.push(2)
print(stack.pop())  # 2

typing常用类型
int、long、float：整型、长整形、浮点型
bool、str：布尔型、字符串类型
List、 Tuple、 Dict、 Set：列表、元组、字典、集合
Iterable、Iterator：可迭代类型、迭代器类型
Generator：生成器类型

python类型指定（类型提示）

python对函数的参数和返回值进行指定类型和检查
https://blog.csdn.net/sgyuanshi/article/details/96192887
https://geek-docs.com/python/python-ask-answer/472_python_how_to_specify_multiple_return_types_using_typehints.html

类型提示是 Python 3.5 引入的一项功能，用于在函数定义中添加类型注释。它可以提供静态类型检查和编辑器自动补全的功能，使得代码更加可读和易于理解。在python3.5之后，就新增了对函数参数和返回值的类型指定和检查，以及在新建变量时也可以指定类型。使用类型提示(Type Hints)来声明类型。

1.指定参数类型

1. 基本类型指定：

def add(x: int, y: int) -> int:return x + y

2. 容器类型

from typing import List, Tuple, Dict
def add(a: int, string: str, f: float,b: bool) -> Tuple[List, Tuple, Dict, bool]:list1 = list(range(a))tup = (string, string, string)d = {"a": f}bl = breturn list1, tup, d, bl
print(add(5, "hhhh", 2.3, False))
([0, 1, 2, 3, 4], ('hhhh', 'hhhh', 'hhhh'), {'a': 2.3}, False)

在传入参数时通过"参数名:类型"的形式声明参数的类型；
返回结果通过"-> 结果类型"的形式声明结果的类型。

3. 泛型
   TypeVar允许创建泛型函数或类。Callable和Sequence等泛型类型的使用。

T = TypeVar('T')  # Can be anything
A = TypeVar('A', str, bytes)  # Must be str or bytes
A = Union[str, None] # Must be str or None

from typing import Sequence, TypeVar, UnionT = TypeVar('T')      # Declare type variabledef first(l: Sequence[T]) -> T:   # Generic functionreturn l[0]

4. Union类型注释
   https://fastapi.tiangolo.com/python-types
   Union 类型允许同时使用多个数据类型，其中任何一种类型的值都可以传递给函数。在注释中，我们使用 or 或 | 分隔多个数据类型。您可以声明变量是几种类型中的任意一种，例如int或str。
   在 Python 3.6 及更高版本（包括 Python 3.10）中，您可以使用类型Unionfromtyping并在方括号内放入可能接受的类型。
   在 Python 3.10 中还有一种新语法，您可以用竖线 ( |)分隔可能的类型。

#Python 3.6 及更高版本
from typing import Union
def process_item(item: Union[int, str]):print(item)
# Python 3.10 中还有一种新语法，您可以用竖线 ( |)分隔可能的类型。
def process_item(item: int | str):print(item)

5. Optional 类型注释
   Optional类型表示一个可选的数据类型，它可用于表示参数可以是一种数据类型或 None 值。我们使用 Optional[type] 表示该函数参数可以是 type 或 None 值。
   使用Optional[str]而不是 只会str让编辑器帮助您检测错误，您可能会假设某个值始终是str，而实际上它也可能也是None。Optional[Something]实际上是 的简写Union[Something, None]，它们是等效的。这也意味着在 Python 3.10 中，你可以使用Something | None：

#Python 3.6 及更高版本
from typing import Optional
def say_hi(name: Optional[str] = None):if name is not None:print(f"Hey {name}!")else:print("Hello World")from typing import Union
def say_hi(name: Union[str, None] = None):if name is not None:print(f"Hey {name}!")else:print("Hello World")#在 Python 3.10 中，你可以使用Something | None：
def say_hi(name: str | None = None):if name is not None:print(f"Hey {name}!")else:print("Hello World")

6. 嵌套的容器类型
   嵌套容器（Nested container）是指容器中又包含其他容器。 typing 模块为这种数据类型提供了更复杂的注释方式。

from typing import List, Tupledef my_function(arg1: List[Tuple[int, str]]) -> List[str]:"""接受一个整型列表中包含元组（整型，字符串），返回由元组中包含的字符串组成的列表。"""return [x[1] for x in arg1]

from pydantic import BaseModelclass PetCls:def __init__(self, *, name: str, species: str):self.name = nameself.species = speciesclass PersonCls:def __init__(self, *, name: str, age: float = None, pets: list[PetCls]):self.name = nameself.age = ageself.pets = petsclass Pet(BaseModel):name: strspecies: strclass Config:orm_mode = Trueclass Person(BaseModel):name: strage: float = Nonepets: list[Pet]class Config:orm_mode = Truebones = PetCls(name='Bones', species='dog')
orion = PetCls(name='Orion', species='cat')
anna = PersonCls(name='Anna', age=20, pets=[bones, orion])
anna_model = Person.from_orm(anna)
print(anna_model)
#> name='Anna' age=20.0 pets=[Pet(name='Bones', species='dog'),
#> Pet(name='Orion', species='cat')]

7. Any 类型
   Any 是一种特殊的类型。静态类型检查器将所有类型视为与 Any 兼容，反之亦然， Any 也与所有类型相兼容。这意味着可对类型为 Any 的值执行任何操作或方法调用，并将其赋值给任何变量。

2、指定返回值类型

返回结果通过"-> 结果类型"的形式声明结果的类型。
如何指定多个返回类型
在 Python 中，单个函数可以返回多个不同类型的值。这种情况下，我们可以使用 typing 模块中的 Union 类来指定多个返回类型。
typing.Union 类接受一个或多个类型作为参数，并返回一个包含了这些类型的联合类型。下面是一个使用 Union 类指定多个返回类型的示例：

from typing import Uniondef divide(x: int, y: int) -> Union[int, float]:if y == 0:return float('inf')else:return x / y

在这个示例中，函数 divide 接受两个参数 x 和 y 的类型都为 int，并且返回类型为 Union[int, float]。如果 y 的值为 0，函数返回 float(‘inf’)，否则返回 x / y 的结果。

3、类型检查工具

在 Python 中，我们可以使用第三方工具进行类型检查，以保证代码的正确性。常用的类型检查工具有 mypy、pylint 和 pyright 等。
mypy 是最受欢迎的静态类型检查工具之一，它可以检查代码中的类型错误和不一致之处，并提供有关如何修复这些错误的建议。使用 mypy 可以使代码更加健壮和可维护。

from typing import Uniondef divide(x: int, y: int) -> Union[int, float]:if y == 0:return float('inf')else:return x / yresult = divide(5, 2)
print(result)

在命令行中运行以下命令进行类型检查：
mypy test.py

4、非空设置（字段）

方法一：（值约束）
Pydantic 中导入 Field设置字段校验，字符窜设置其长度大于1.min_length=1.
https://docs.pydantic.dev/latest/concepts/fields/#string-constraints

from pydantic import BaseModel, Fieldclass Foo(BaseModel):short: str = Field(min_length=3)long: str = Field(max_length=10)regex: str = Field(pattern=r'^\d*$')  foo = Foo(short='foo', long='foobarbaz', regex='123')
print(foo)
#> short='foo' long='foobarbaz' regex='123'

方法二：（键约束，将空的键值直接删除），只能排除键，不能排除值
排除，该参数可用于控制应从 模型。exclude
https://docs.pydantic.dev/latest/concepts/serialization/#advanced-include-and-exclude
https://docs.pydantic.dev/latest/concepts/serialization/#modelmodel_dump

from typing import Optionalfrom pydantic import BaseModel, Fieldclass Person(BaseModel):name: strage: Optional[int] = Field(None, exclude=False)person = Person(name='Jeremy')print(person.model_dump())
#> {'name': 'Jeremy', 'age': None}
print(person.model_dump(exclude_none=True))  #去除值为空键值
#> {'name': 'Jeremy'}
print(person.model_dump(exclude_unset=True))  
#> {'name': 'Jeremy'}
print(person.model_dump(exclude_defaults=True))  
#> {'name': 'Jeremy'}

from typing import Any, List, Optionalfrom pydantic import BaseModel, Field, Jsonclass BarModel(BaseModel):whatever: intclass FooBarModel(BaseModel):banana: Optional[float] = 1.1foo: str = Field(serialization_alias='foo_alias')bar: BarModelm = FooBarModel(banana=3.14, foo='hello', bar={'whatever': 123})# returns a dictionary:
print(m.model_dump())
#> {'banana': 3.14, 'foo': 'hello', 'bar': {'whatever': 123}}
print(m.model_dump(include={'foo', 'bar'}))
#> {'foo': 'hello', 'bar': {'whatever': 123}}
print(m.model_dump(exclude={'foo', 'bar'}))
#> {'banana': 3.14}
print(m.model_dump(by_alias=True))
#> {'banana': 3.14, 'foo_alias': 'hello', 'bar': {'whatever': 123}}
print(FooBarModel(foo='hello', bar={'whatever': 123}).model_dump(exclude_unset=True)
)
#> {'foo': 'hello', 'bar': {'whatever': 123}}
print(FooBarModel(banana=1.1, foo='hello', bar={'whatever': 123}).model_dump(exclude_defaults=True)
)
#> {'foo': 'hello', 'bar': {'whatever': 123}}
print(FooBarModel(foo='hello', bar={'whatever': 123}).model_dump(exclude_defaults=True)
)
#> {'foo': 'hello', 'bar': {'whatever': 123}}
print(FooBarModel(banana=None, foo='hello', bar={'whatever': 123}).model_dump(exclude_none=True)
)
#> {'foo': 'hello', 'bar': {'whatever': 123}}class Model(BaseModel):x: List[Json[Any]]print(Model(x=['{"a": 1}', '[1, 2]']).model_dump())
#> {'x': [{'a': 1}, [1, 2]]}
print(Model(x=['{"a": 1}', '[1, 2]']).model_dump(round_trip=True))
#> {'x': ['{"a":1}', '[1,2]']}

5、Literal只能包含哪些值（某个键只能对应哪些值）

https://docs.pydantic.dev/1.10/usage/types/#literal-type

from typing import Literalfrom pydantic import BaseModel, ValidationErrorclass Pie(BaseModel):flavor: Literal['apple', 'pumpkin']Pie(flavor='apple')
Pie(flavor='pumpkin')
try:Pie(flavor='cherry')
except ValidationError as e:print(str(e))"""1 validation error for Pieflavorunexpected value; permitted: 'apple', 'pumpkin'(type=value_error.const; given=cherry; permitted=('apple', 'pumpkin'))

6、模型配置（指定全局字段）

https://hellowac.github.io/pydantic-zh-cn/v1.10.7-zh-cn/usage/model_config/

模型配置
pydantic 的行为可以通过模型上的 Config 类或 pydantic 数据类来控制。Python 3.7 及以上版本from pydantic import BaseModel, ValidationErrorclass Model(BaseModel):v: strclass Config:max_anystr_length = 10error_msg_templates = {'value_error.any_str.max_length': 'max_length:{limit_value}',}try:Model(v='x' * 20)
except ValidationError as e:print(e)"""1 validation error for Modelvmax_length:10 (type=value_error.any_str.max_length; limit_value=10)"""

6、请求示例

from typing import Unionfrom fastapi import FastAPI
import uvicorn
from pydantic import BaseModel,Field
from typing import Union,Optional,Literalapp = FastAPI()# 1、get请求@app.get("/")
def read_root():return {"Hello": "World"}#  http://127.0.0.1:8000/items/5?q=somequery
@app.get("/items1/{item_id}")
def read_item(item_id: int, q: Union[str, None] = None):return {"item_id": item_id, "q": q}fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
# http://127.0.0.1:8000/items/?skip=0&limit=10
@app.get("/items2/")
async def read_item(skip: int = 0, limit: int = 10):return fake_items_db[skip : skip + limit]#http://127.0.0.1:8000/items/foo?short=1
# http://127.0.0.1:8000/items/foo?short=True,q=dotest
@app.get("/items3/{item_id}")
async def read_item(item_id: str, q: str | None = None, short: bool = False):item = {"item_id": item_id}if q:item.update({"q": q})if not short:item.update({"description": "This is an amazing item that has a long description"})return item# 2、post请求
#与声明查询参数时相同，当模型属性具有默认值时，则不需要它。
#否则是需要的。使用 None 使其成为可选的。
'''
{"name": "1","include":"apple","name1": "","name2": "","description": "An optional description","price": 45.2,"tax": 3.5
}
'''class Item(BaseModel):name: str=Field(min_length=1) #非空# name: str=Field(None, exclude=True) #去除空的键include: Literal['apple', 'pumpkin'] #只能包含哪些值name1:Optional[str] #选择，可以为空值，带none时是指键可以没有name2:Union[str]   #选择，可以为空，带none时是指键可以没有description: str | None = None #Union可以用|替代，python10price: floattax: float | None = None# @app.post("/items/")
# async def create_item(item: Item):
#     return item@app.post("/post_items1/")
async def create_item(item: Item):item_dict = item.dict()if item.tax:price_with_tax = item.price + item.taxitem_dict.update({"price_with_tax": price_with_tax})return item_dict# 3、请求体嵌套
'''
{"name": "Foo","description": "The pretender","price": 42.0,"tax": 3.2,"tags": ["rock", "metal", "bar"],"image": {"url": "http://example.com/baz.jpg","name": "The Foo live"}
}
'''class Image(BaseModel):url: strname: strclass Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Nonetags: set[str] = set()image: Image | None = None@app.post("/post_items2/{item_id}")
async def update_item(item_id: int, item: Item):print('test')results = {"item_id": item_id, "item": item}return results
# 传入多个body
'''
{"item": {"name": "Foo","description": "The pretender","price": 42.0,"tax": 3.2},"user": {"username": "dave","full_name": "Dave Grohl"}
}
'''
class Item3(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Noneclass User3(BaseModel):username: strfull_name: str | None = None@app.post("/post_items3/{item_id}")
async def update_item(item_id: int, item: Item3, user: User3):results = {"item_id": item_id, "item": item, "user": user}return results# 4、数据校验
# Query:校验get请求参数
# Path:校验路径参数
# Body:校验post请求体
# Field:校验 Pydantic 模型内部声明校验和元数据if __name__ == "__main__":#方法一：# config = uvicorn.Config("main:app", host='0.0.0.0',port=8888, reload=True, log_level="info") #指定模块，当前用FastAPI()的appconfig = uvicorn.Config('getpost:app',host='0.0.0.0',port=8888, reload=True, log_level="info")# config = uvicorn.Config(app,host='0.0.0.0',port=8888, reload=True, log_level="info") #用app，只有在不使用多处理（worker=NUM）或重新加载（reload=True）的情况下，此样式才有效，因此我们建议使用导入字符串样式server = uvicorn.Server(config)server.run()#方法二：# from pathlib import Path# uvicorn.run('getpost:app',host='0.0.0.0',port=8888, reload=True, log_level="info")

fastapi_727">7、fastapi请求数据异常捕获

https://fastapi.tiangolo.com/zh/tutorial/handling-errors/?h=
如在调用路径操作函数里的工具函数时，触发了 HTTPException，FastAPI 就不再继续执行路径操作函数中的后续代码，而是立即终止请求，并把 HTTPException 的 HTTP 错误发送至客户端。在介绍依赖项与安全的章节中，您可以了解更多用 raise 异常代替 return 值的优势。

覆盖默认异常处理器：
FastAPI 自带了一些默认异常处理器。
触发 HTTPException 或请求无效数据时，这些处理器返回默认的 JSON 响应结果。不过，也可以使用自定义处理器覆盖默认异常处理器。请求中包含无效数据时，FastAPI 内部会触发 RequestValidationError。该异常也内置了默认异常处理器。覆盖默认异常处理器时需要导入 RequestValidationError，并用 @app.excption_handler(RequestValidationError) 装饰异常处理器。这样，异常处理器就可以接收 Request 与异常。

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException
from pydantic import BaseModel
from fastapi.responses import JSONResponse
import uvicorn
import sys
app = FastAPI()@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):print('内部异常:')print('请求路径:',request.url,request.base_url)#获取请求体if request.method == "POST":try:   json_body = await request.json()print('post请求体:',json_body)except Exception as e:# print('json处理异常:',Exception)print(f"Exception type: {type(e).__name__}")print(f"Exception message: {e}")print(f"Exception details: {sys.exc_info()}")return JSONResponse(status_code=418,content={"message": '请求json解析异常,json可能为空或不是json格式'},)if request.method == "GET":try:   # json_body = await request.json()json_body = dict(request.query_params)  # 将查询参数转换为普通字典print('get请求体:',json_body)except Exception as e:# print('json处理异常:',Exception)print(f"Exception type: {type(e).__name__}")print(f"Exception message: {e}")print(f"Exception details: {sys.exc_info()}")return JSONResponse(status_code=418,content={"message": '请求json解析异常,json可能为空或不是json格式'},)print('抛出异常信息:')print(exc)#内部异常可能没有errors和body属性，不用两项# print("detail",exc.errors())# print("body",exc.body)return PlainTextResponse(str('捕获到内部异常：'+str(exc)), status_code=exc.status_code)# return PlainTextResponse(str(exc.detail), status_code=exc.status_code)@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):print('请求体异常:')print('请求路径:',request.url,request.base_url)#request异常有errors和body属性，可以获取到请求体print('抛出异常信息:')print(exc)print("detail",exc.errors())print("body",exc.body) #请求体return PlainTextResponse(str(exc), status_code=400)@app.get("/items/{item_id}")
async def read_item(item_id: int):if item_id == 3:raise HTTPException(status_code=418, detail="Nope! I don't like 3.") #模拟内部抛出异常return {"item_id": item_id}class Item(BaseModel):title: strsize: int@app.post("/items/")
async def create_item(item: Item):return itemif __name__ == "__main__":#方法一：# config = uvicorn.Config("main:app", host='0.0.0.0',port=8888, reload=True, log_level="info") #指定模块，当前用FastAPI()的app# config = uvicorn.Config('getpost:app',host='0.0.0.0',port=8888, reload=True, log_level="info")config = uvicorn.Config(app,host='0.0.0.0',port=8888, reload=True, log_level="info") #用app，只有在不使用多处理（worker=NUM）或重新加载（reload=True）的情况下，此样式才有效，因此我们建议使用导入字符串样式server = uvicorn.Server(config)server.run()#方法二：# from pathlib import Path# uvicorn.run('getpost:app',host='0.0.0.0',port=8888, reload=True, log_level="info")

请求信息：查看request类
https://fastapi.tiangolo.com/zh/reference/request/#fastapi.Request.url
比如路径：

	print('请求体异常:')print('请求路径:',request.url,request.base_url)json_body = await request.json()print('请求体:',json_body)# print('请求体:',request.json()) #异步不行

8、多进程部署

uvicorn文档：https://www.uvicorn.org/deployment/
fastapi文档：https://fastapi.tiangolo.com/deployment/
Gunicorn 不支持直接传递应用程序参数，所以通常需要通过环境变量或配置文件来实现。不支持argparse。日志相关看下面的日志。
–reload 和 --workers 参数是互斥的。

1.代码：代码中设置多进程，以编程方式运行
https://www.uvicorn.org/deployment/#running-programmatically
要直接从 Python 程序中运行，您应该使用 uvicorn.run（app， **config）。请注意：传递应用程序实例本身，而不是应用程序导入字符串，uvicorn.run(app, host="127.0.0.1", port=5000, log_level="info")，此样式仅在您不使用 multiprocessing （workers=NUM） 或重新加载 （reload=True） 时才有效，因此我们建议使用 import string 样式。还要注意，在这种情况下，你应该把 uvicorn.run 放在主模块的 if == ‘main’ 子句中__name__。reload 和 workers 参数是互斥的。

官方推荐：gunicorn main:app --workers 4 --worker-class uvicorn.workers.UvicornWorker --bind 0.0.0.0:80

#llama_emb_server为文件名
import uvicorn
from fastapi import FastAPI
app = FastAPI()
# get请求，健康测试
@app.get("/")
async def healthcheck():print('健康检测')return {"status": "healthy"}
if __name__ == "__main__":
#llama_emb_server为本代码文件名
# 1、测试启动方法，重新加载reload。reload 和 workers 参数是互斥的。#方法一：config# config = uvicorn.Config("main:app", host='0.0.0.0',port=8888, reload=True, log_level="info") #指定模块，当前用FastAPI()的app# 测试:reload=True调试# config = uvicorn.Config(app,host='0.0.0.0',port=port, reload=True, log_level="info") #用app，只有在不使用多处理（worker=NUM）或重新加载（reload=True）的情况下，此样式才有效，因此我们建议使用导入字符串样式# 生产：reload 和 workers 参数是互斥的。config方法不生效，必须用run# config = uvicorn.Config(app,host='0.0.0.0',port=port,reload=False, workers=4, log_level="info") #此方法不生效# server = uvicorn.Server(config)# server.run()#方法二：run# uvicorn.run("llama_emb_server:app", host="0.0.0.0", port=8888,reload=True,  log_level="info")#其他方法：FastAPI CLI启动，参考上面的启动方式# 2、生产启动方法，多进程works。reload 和 workers 参数是互斥的。# 方法一：传workers=NUM参数，运行例如：python llama_emb_server.py# uvicorn.run("llama_emb_server:app", host="127.0.0.1", port=5000,workers=4,  log_level="info")uvicorn.run("llama_emb_server:app", host="0.0.0.0", port=port,workers=4,  log_level="info")# 方法二：用WEB_CONCURRENCY环境变量，运行例如：WEB_CONCURRENCY=6 python llama_emb_server.py# uvicorn.run("llama_emb_server:app", host="0.0.0.0", port=port, log_level="info")# 方法三：命令行方式# uvicorn main:app --host 0.0.0.0 --port 8080 --workers 4# gunicorn main:app --workers 4 --worker-class uvicorn.workers.UvicornWorker --bind 0.0.0.0:80# nohup gunicorn -w 5 -k uvicorn.workers.UvicornWorker --timeout 600 main:app --bind 0.0.0.0:8888  --log-level debug --access-logfile ../logs/gunicorn_output.log >> ./logs/log_scure.log 2>&1 &#推荐使用
测试：uvicorn.run("llama_emb_server:app", host="0.0.0.0", port=8888,reload=True,  log_level="info")
生产：uvicorn.run("llama_emb_server:app", host="127.0.0.1", port=5000,workers=4,  log_level="info")

9、应用案例：llama cpu实现文本转embeding服务

环境安装参考：https://blog.csdn.net/weixin_44986037/article/details/141465479

from typing import Union
from fastapi import FastAPI,Query
import uvicorn
from pydantic import BaseModel,Field
from typing import Union,Optional,Literal
import argparse
import os
from typing import (Deque
)
#低版本python，3.10以下
# from typing import (
#     Deque, Dict, FrozenSet, List, Optional, Sequence, Set, Tuple, Union
# )#异常相关包
from fastapi import HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException
from fastapi.responses import JSONResponse
import sysapp = FastAPI()#1、全局异常捕获
#代码内部异常
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):print('*'*100)uid=str(uuid.uuid1())print("request_id请求ID:",uid)print('内部异常:')print('请求路径:',request.method,request.url)#获取请求体if request.method == "POST":try:   json_body = await request.json()print('post请求体:',json_body)except Exception as e:# print('json处理异常:',Exception)print('post请求json解析异常,json可能为空或不是json格式')print(f"Exception type: {type(e).__name__}")print(f"Exception message: {e}")print(f"Exception details: {sys.exc_info()}")if request.method == "GET":try:   # json_body = await request.json()json_body = dict(request.query_params)  # 将查询参数转换为普通字典print('get请求体:',json_body)except Exception as e:# print('json处理异常:',Exception)print('get请求解析异常,参数可能为空或格式错误')print(f"Exception type: {type(e).__name__}")print(f"Exception message: {e}")print(f"Exception details: {sys.exc_info()}")print('抛出异常信息:')print(exc)print('*'*100)  return JSONResponse(status_code=500,content={"request_id": uid,"code": "500","message": "server_error","data":[]},)#请参数异常
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):print('*'*100)uid=str(uuid.uuid1())print("request_id请求ID:",uid)print('请求体异常:')print('请求路径:',request.method,request.url)#request异常有errors和body属性，可以获取到请求体print('抛出异常信息:')print(exc)print("detail",exc.errors())print("body",exc.body)print('*'*100)return JSONResponse(status_code=400,content={"request_id": uid,"code": "400","message": "illegal_params","data":[]},)# 2、get请求，健康测试
@app.get("/healthcheck")
async def healthcheck():print('健康检测')return {"status": "healthy"}# from uuid import UUID
import uuid
#3、数据类
# 请求数据
class Request_emb(BaseModel):input:Union[str,list[str],None] #选择，可以为空type:Union[str,None]=None #等价Optional[str]# 返回数据
class Respond_emb(BaseModel):class DataItem(BaseModel):embedding: list[float]index: intrequest_id: strcode: strmessage: strdata: list[DataItem] |None=Nonedata_req={"input":["输入测试1","输入测试2"]
}data_resp={"request_id":'udi213219',"code": "1100","message": "success","data":[{"embedding":[0.004577230662107468, 0.02443666011095047, 0.0027657144237309694],"index": 0}]
}#4、对创建的数据模型进行验证
from pydantic import BaseModel, ValidationError,TypeAdapter
try:# SecureRespond.model_validate(data) #验证输入数据Request_emb.model_validate(data_req) #验证输入数据Respond_emb.model_validate(data_resp) #验证输入数据# print(Request_emb(**data_req).model_dump()) #dict转该数据，转json# print(Respond_emb(**data_resp).model_dump()) #dict转该数据，转jsonprint('请求书验证成功')
except ValidationError as exc:print('异常：')print(repr)# quit('测试')#5、embedding模型数据接口,type=secu时为安全emb，默认为emb
@app.post("/v1/embeddings")  #返回数据校验没生效
async def create_secure(secu_data: Request_emb) :# print('输入：',secu_data.dict())uid=str(uuid.uuid1())print("request_id请求ID:",uid)data_dict = secu_data.dict()print('输入转成字典：',data_dict,type(data_dict))if secu_data.input:print('数据不为空：',secu_data.input)# 检查数据类型是否为字符串if isinstance(secu_data.input, str):# 如果是字符串，将其转换为包含该字符串的列表secu_data.input = [secu_data.input]#Bert模型批量分类data_allbatch=[]batch_size=2 #对数据分批处理for key,bat_data in enumerate([secu_data.input[i:i + batch_size] for i in range(0, len(secu_data.input), batch_size)]):tag=key*batch_sizeprint('标记：',key*batch_size)print(f'第{key}个',bat_data)if secu_data.type =='secu': data_batch=secu_execute(model_security,classify_head,bat_data)else:data_batch=emb_execute(model_emb,bat_data)data_batch=[{"embedding":item,"index":idx+tag} for idx,item in enumerate(data_batch)]data_allbatch.extend(data_batch)#封装所有批数据print('所有批数据：',data_allbatch)# import numpy as np# array = np.array(data_allbatch)# print(array.shape)data_outs={"request_id": uid,"code": "200","message": "success","data":data_allbatch}print('emb模型处理后的数据：',data_outs)else:print('数据为空')return {"request_id": uid,"code": "200","message": "data empty","data":[]}# 校验数据try:data_outs=Respond_emb(**data_outs)  print('校验后数据：',data_outs)  except ValidationError as e:print('校验数据异常：',e)return {"request_id": uid,"code": "500","message": "server_error","data":[]}return data_outsfrom llama_cpp import Llama
import torch
import torch.nn as nnclass BertPooler(nn.Module):def __init__(self):super().__init__()self.dense = nn.Linear(in_features=768, out_features=768, bias=True)self.activation = nn.Tanh()def forward(self, hidden_states: torch.Tensor) -> torch.Tensor:first_token_tensor = hidden_states[:, 0]pooled_output = self.dense(first_token_tensor)pooled_output = self.activation(pooled_output)return pooled_outputclass ClassifyHead(nn.Module):def __init__(self):super().__init__()self.pooler = BertPooler()self.dropout = nn.Dropout(p=0.1, inplace=False)self.classifier = nn.Linear(in_features=768, out_features=6, bias=True)self.sigmoid = nn.Sigmoid()def forward(self, encoder_outputs: torch.Tensor) -> torch.Tensor:pooled_output = self.pooler(encoder_outputs)pooled_output = self.dropout(pooled_output)logits = self.classifier(pooled_output)logits = self.sigmoid(logits)return logits#环境变量获取
Emb_env = os.getenv('Emb_env')
Emb_sucu_env= os.getenv('Emb_sucu_env')
Class_env=os.getenv('Class_env')
Port_env= os.getenv('Port_env')
print('环境变量：',Emb_env,Emb_sucu_env,Port_env)
port=8888
if Port_env:port=Port_env 
#加载embedding模型
model_emb_path = "XXX/models/bge-base-zh-v1.5-gguf/bge-base-zh-v1.5-q8_0.gguf"
if Emb_env:model_emb_path=Emb_env
model_emb = Llama(model_emb_path, embedding=True)def emb_execute(model,str_list):### embedding model #### print('emb文本：',str_list)# embeddings = model.embed(str_list)  # print('编码后：',type(embeddings),embeddings)return model.embed(str_list) #加载安全模型
model_security_path = "XXX/models/security_0.1b_v0.04.gguf"
if Emb_sucu_env:model_security_path=Emb_sucu_env
model_security = Llama(model_security_path, embedding=True)classify_head = ClassifyHead()
class_path="XXX/model/classify_head.pt"
if Class_env : class_path=Class_env
classify_head.load_state_dict(torch.load(class_path)) def secu_execute(model_security,classify_head,str_list):outputs = model_security.embed(str_list)all_embs=[]for out in outputs:output1 = torch.tensor([out])with torch.no_grad():scores1 = classify_head(output1).tolist()all_embs.extend(scores1)# print('secu编码后：',all_embs)return all_embsprint('测试进程')if __name__ == "__main__":# 1、测试启动方法，重新加载reload。reload 和 workers 参数是互斥的。#方法一：config# config = uvicorn.Config("main:app", host='0.0.0.0',port=8888, reload=True, log_level="info") #指定模块，当前用FastAPI()的app# 测试:reload=True调试# config = uvicorn.Config(app,host='0.0.0.0',port=port, reload=True, log_level="info") #用app，只有在不使用多处理（worker=NUM）或重新加载（reload=True）的情况下，此样式才有效，因此我们建议使用导入字符串样式# 生产：reload 和 workers 参数是互斥的。config方法不生效，必须用run# config = uvicorn.Config(app,host='0.0.0.0',port=port,reload=False, workers=4, log_level="info") #此方法不生效# server = uvicorn.Server(config)# server.run()#方法二：run# uvicorn.run("llama_emb_server:app", host="0.0.0.0", port=8888,reload=True,  log_level="info")# 2、生产启动方法，多进程works。reload 和 workers 参数是互斥的。# 方法一：传workers=NUM参数，运行例如：python llama_emb_server.py# uvicorn.run("llama_emb_server:app", host="127.0.0.1", port=5000,workers=4,  log_level="info")uvicorn.run("llama_emb_server:app", host="0.0.0.0", port=port,workers=4,  log_level="info")# 方法二：用WEB_CONCURRENCY环境变量，运行例如：WEB_CONCURRENCY=6 python llama_emb_server.py# uvicorn.run("llama_emb_server:app", host="0.0.0.0", port=port, log_level="info")# 方法三：命令行方式# uvicorn main:app --host 0.0.0.0 --port 8080 --workers 4# gunicorn main:app --workers 4 --worker-class uvicorn.workers.UvicornWorker --bind 0.0.0.0:80# nohup gunicorn -w 5 -k uvicorn.workers.UvicornWorker --timeout 600 main:app --bind 0.0.0.0:8888  --log-level debug --access-logfile ../logs/gunicorn_output.log >> ./logs/log_scure.log 2>&1 &#   uvicorn workers Running programmatically
# uvicorn llama_emb_server:app --host 0.0.0.0 --port 8080 --workers 4
# WEB_CONCURRENCY=6 python llama_emb_server.py

使用说明：

#默认使用emb模型，"type":"secu"时用secure emb模型
curl --location 'http://172.31.208.3:8077/v1/embeddings' \
--header 'Content-Type: application/json' \
--data '{
"input":["输入测试1","输入测试2","in3","in4","in5"]
}'#"type":"secu"时用secure emb模型
curl --location 'http://172.31.208.3:8077/v1/embeddings' \
--header 'Content-Type: application/json' \
--data '{
"input":["输入测试1","输入测试2","in3","in4","in5"],
"type":"secu"
}'# 请求和返回数据格式
data_req={"input":["输入测试1","输入测试2"],"type":""
}data_resp={"request_id":'udi213219',"code": "1100","message": "success","data":[{"embedding":[0.004577230662107468, 0.02443666011095047, 0.0027657144237309694],"index": 0}]
}

10、多进程日志（FastApi结合loguru日志使用）

在多进程环境中，可能会遇到一些与日志处理相关的问题。考虑使用其他日志库，如loguru、structlog等，它们提供了更好的多进程支持和灵活的配置选项。
FastApi结合loguru日志使用，多进程：https://blog.csdn.net/qq_51967017/article/details/134045236
1.log.py文件

import os
import sys
import time
import logging
from types import FrameType
from typing import cast
from loguru import logger
# from path_conf import LogPath
LogPath='./logts'class Logger:"""输出日志到文件和控制台"""def __init__(self):# 文件的命名log_name = f"Fast_{time.strftime('%Y-%m-%d', time.localtime()).replace('-', '_')}.log"log_path = os.path.join(LogPath, "Fast_{time:YYYY-MM-DD}.log")self.logger = logger# 清空所有设置self.logger.remove()# 判断日志文件夹是否存在，不存则创建if not os.path.exists(LogPath):os.makedirs(LogPath)# 日志输出格式formatter = "{time:YYYY-MM-DD HH:mm:ss} | {level}: {message}"# 添加控制台输出的格式,sys.stdout为输出到屏幕;关于这些配置还需要自定义请移步官网查看相关参数说明self.logger.add(sys.stdout,format="<green>{time:YYYYMMDD HH:mm:ss}</green> | "  # 颜色>时间"{process.name} | "  # 进程名"{thread.name} | "  # 进程名" PID: {process}|""<cyan>{module}</cyan>.<cyan>{function}</cyan>"  # 模块名.方法名":<cyan>{line}</cyan> | "  # 行号"<level>{level}</level>: "  # 等级"<level>{message}</level>",  # 日志内容)# 日志写入文件self.logger.add(log_path,  # 写入目录指定文件format='{time:YYYYMMDD HH:mm:ss} - '  # 时间"{process.name} | "  # 进程名"{thread.name} | "  # 进程名" PID: {process}|"'{module}.{function}:{line} - {level} -{message}',  # 模块名.方法名:行号encoding='utf-8',retention='7 days',  # 设置历史保留时长backtrace=True,  # 回溯diagnose=True,  # 诊断enqueue=True,  # 异步写入rotation="00:00",  # 每日更新时间# rotation="5kb",  # 切割，设置文件大小，rotation="12:00"，rotation="1 week"# filter="my_module"  # 过滤模块# compression="zip"   # 文件压缩)def init_config(self):LOGGER_NAMES = ("uvicorn.asgi", "uvicorn.access", "uvicorn")# change handler for default uvicorn loggerlogging.getLogger().handlers = [InterceptHandler()]for logger_name in LOGGER_NAMES:logging_logger = logging.getLogger(logger_name)logging_logger.handlers = [InterceptHandler()]def get_logger(self):return self.loggerclass InterceptHandler(logging.Handler):def emit(self, record: logging.LogRecord) -> None:  # pragma: no cover# Get corresponding Loguru level if it existstry:level = logger.level(record.levelname).nameexcept ValueError:level = str(record.levelno)# Find caller from where originated the logged messageframe, depth = logging.currentframe(), 2while frame.f_code.co_filename == logging.__file__:  # noqa: WPS609frame = cast(FrameType, frame.f_back)depth += 1logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage(),)Loggers = Logger()
log = Loggers.get_logger()

2.main.py文件

import uvicorn
from fastapi import FastAPI
from log import log, Loggers
import time
from pydantic import BaseModel# # 延时 5 秒
# time.sleep(5)app = FastAPI()log.info(f"【start】secuer日志记录器创建成功")@app.get("/")
def index():# log.error("/index")log.info(f"这是一个get请求")# 延时 3 秒time.sleep(3)return "Hello, World."class Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = None@app.post("/items/")
async def create_item(item: Item):# 延时 3 秒time.sleep(3)log.info(f"这是一个post请求：{item}")return itemif __name__ == '__main__':# config = uvicorn.Config("main:app", host='0.0.0.0', port=9999)uvicorn.run("main:app", host="0.0.0.0", port=8888,workers=4,  log_level="info")server = uvicorn.Server(config)# 将uvicorn输出的全部让loguru管理Loggers.init_config()server.run()

3.测压press_test.py

import asyncio
import aiohttpasync def fetch_url(session, url):async with session.get(url) as response:status_code = response.statuscontent = await response.text()return status_code, content[:100]  # 只返回状态码和前100字符的响应内容async def main():url = "http://172.31.208.3:8077"async with aiohttp.ClientSession() as session:tasks = [fetch_url(session, url) for _ in range(100)]results = await asyncio.gather(*tasks)for status_code, content in results:print(f"Status: {status_code}, Content: {content}")# 运行异步任务
asyncio.run(main())

cx_Freeze打包，中文编码对log的影响

cx_Freeze打包中文编码与logger冲突，打包的时候要注释掉logger,放开中文编码注释
多进程启动：uvicorn demo11test:app --host 0.0.0.0 --port 8888 --workers 4 时，
报错：logging stream.write(msg + self.terminator) ValueError: underlying buffer has been detached
原因：cx_Freeze打包时，解决编码问题的代码影响。

检查流对象是否被关闭：
确保你在程序运行过程中没有意外关闭或分离流对象。
import sys
# 确保 sys.stdout 或 sys.stderr 没有被关闭
assert not sys.stdout.closed
assert not sys.stderr.closed

解决：将以下代码注释掉

# # 解决打包中文编码问题
# import sys
# import codecs
# # 确保标准输出和错误输出使用UTF-8编码
# sys.stdout = codecs.getwriter("utf-8")(sys.stdout.detach())
# sys.stderr = codecs.getwriter("utf-8")(sys.stderr.detach())

11、文件加载

参考：CosyVoice fastapi
https://github.com/FunAudioLLM/CosyVoice

# Set inference model
# export MODEL_DIR=pretrained_models/CosyVoice-300M-Instruct
# For development
# fastapi dev --port 6006 fastapi_server.py
# For production deployment
# fastapi run --port 6006 fastapi_server.pyimport os
import sys
import io,time
from fastapi import FastAPI, Response, File, UploadFile, Form
from fastapi.responses import HTMLResponse
from fastapi.middleware.cors import CORSMiddleware  #引入 CORS中间件模块
from contextlib import asynccontextmanager
ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.append('{}/../../..'.format(ROOT_DIR))
sys.path.append('{}/../../../third_party/Matcha-TTS'.format(ROOT_DIR))
from cosyvoice.cli.cosyvoice import CosyVoice
from cosyvoice.utils.file_utils import load_wav
import numpy as np
import torch
import torchaudio
import logging
logging.getLogger('matplotlib').setLevel(logging.WARNING)class LaunchFailed(Exception):pass@asynccontextmanager
async def lifespan(app: FastAPI):model_dir = os.getenv("MODEL_DIR", "pretrained_models/CosyVoice-300M-SFT")if model_dir:logging.info("MODEL_DIR is {}", model_dir)app.cosyvoice = CosyVoice(model_dir)# sft usagelogging.info("Avaliable speakers {}", app.cosyvoice.list_avaliable_spks())else:raise LaunchFailed("MODEL_DIR environment must set")yieldapp = FastAPI(lifespan=lifespan)#设置允许访问的域名
origins = ["*"]  #"*"，即为所有,也可以改为允许的特定ip。
app.add_middleware(CORSMiddleware, allow_origins=origins,  #设置允许的origins来源allow_credentials=True,allow_methods=["*"],  # 设置允许跨域的http方法，比如 get、post、put等。allow_headers=["*"])  #允许跨域的headers，可以用来鉴别来源等作用。def buildResponse(output):buffer = io.BytesIO()torchaudio.save(buffer, output, 22050, format="wav")buffer.seek(0)return Response(content=buffer.read(-1), media_type="audio/wav")@app.post("/api/inference/sft")
@app.get("/api/inference/sft")
async def sft(tts: str = Form(), role: str = Form()):start = time.process_time()output = app.cosyvoice.inference_sft(tts, role)end = time.process_time()logging.info("infer time is {} seconds", end-start)return buildResponse(output['tts_speech'])