一.基本介绍

1.什么是 fastapi?

2.有什么特点?

二.文件相关

1.客户端

@app.post("/file-to-text")
async def file_to_text(
    request: Request,
    file: UploadFile = Form(...),
    api: str = Form(...),
):
    try:
        # 构建文件和普通参数的表单数据
        files = {"file": (file.filename, await file.read(), file.content_type)}
        data = {"api": api}
        async with httpx.AsyncClient(timeout=timeout) as client:
            response = await client.post(
                f"/api/file-to-text",
                files=files,
                data=data,
            )
            response.raise_for_status()
            return response.json()
    except httpx.ReadTimeout:
        return {"code": -1, "message": "请求超时"}

2.服务端

@staticmethod
@Voice2TextRouter.post("")
async def file_to_text(
        file: Annotated[bytes, File()],
        api: str = Form(...),
) -> dict:
     return await Voice2TextRouterMap.file_to_text(api=api, file=file)

1.Query 用法

这段代码是使用 FastAPI 框架中的 Query 参数定义方式。让我来解释它：

name: str = Query(None)

这行代码定义了一个名为 name 的参数，它是一个字符串类型的参数。参数的默认值是 None。这个参数使用了 FastAPI 提供的 Query 类来定义，用于从查询字符串中获取参数值。

在 FastAPI 中，Query 类用于指定参数的一些元数据，例如默认值、描述、参数是否必需等。在这个例子中，我们只指定了参数的默认值为 None。这意味着如果请求中没有提供 name 参数，它将被视为 None。

这样的定义使得 name 参数成为可选的，因为它的默认值是 None。如果需要将参数设置为必需，可以通过指定第一个参数为参数的默认值，然后使用 ... 作为占位符来指示它是必需的，例如 name: str = Query(...)。

1.@staticmethod

在 Python 中，@staticmethod 是一个装饰器，用于定义静态方法。静态方法是指在类中定义的方法，它与类的实例无关，不需要访问类的属性或实例属性，因此可以在不创建类的实例的情况下直接通过类名来调用。

使用 @staticmethod 装饰器可以将一个方法声明为静态方法。静态方法的定义与普通方法类似，但是在方法定义上方加上 @staticmethod 装饰器。

class MyClass:
    @staticmethod
    def static_method():
        print("This is a static method")

# 调用静态方法
MyClass.static_method()  # 输出: This is a static method

在静态方法中，没有 self 参数，因为静态方法不会被实例化的对象所调用。静态方法可以在类定义的外部直接调用，也可以通过类名来调用。

静态方法通常用于不需要访问实例属性或类属性的方法，而只是为了组织代码，将相关的功能放在类的命名空间下。静态方法可以提高代码的可读性和可维护性，并且可以在不创建类的实例的情况下直接调用。

2. @KBRouter_old

在 FastAPI 中，@KBRouter_old 和 @KBRouter 是两个不同的路由器类或装饰器，用于定义 API 的端点。通常，这两者之间的区别取决于它们的具体实现和用途，以及在你的应用程序中如何使用它们。

以下是可能的区别：

1. 功能或行为：@KBRouter_old 和 @KBRouter 可能具有不同的功能或行为。它们可能会应用不同的中间件、验证、权限控制或其他功能，以满足特定的需求。

2. 版本或迭代：@KBRouter_old 可能是旧版本的路由器，而 @KBRouter 则是新版本。新版本可能添加了新的功能、修复了 bug 或者改进了性能。

3. 用途：它们可能被用于不同的部分或者模块。例如，@KBRouter_old 可能用于处理旧版本的 API 请求，而 @KBRouter 则处理最新版本的请求。

4. 命名习惯：有时候，开发人员可能使用类似 @Old 或 @Deprecated 的命名惯例来指示某些功能已经过时，但仍然存在于代码中以保持向后兼容性。

3.类定义

class Collection(BaseModel):
    id: Optional[str] = None
    name: str
    metadata: Dict = {}

使用 Pydantic 库定义了一个名为 Collection 的数据模型（Data Model），该模型继承自 BaseModel。Pydantic 是一个用于数据验证和序列化的 Python 库，它通过简单的方式定义数据模型，并提供了自动验证和转换输入数据的功能。

让我们来解释这个数据模型的各个部分：

1. class Collection(BaseModel):: 这定义了一个名为 Collection 的类，它是 BaseModel 的子类。这意味着 Collection 类会继承 BaseModel 类的所有特性和方法，包括数据验证、序列化和反序列化等功能。

2. id: Optional[str] = None: 这是一个属性声明，表示 Collection 类中的一个字段。id 是一个可选的字符串类型，通过 Optional[str] 定义。Optional 表示该字段可以为空值，str 表示该字段的类型是字符串。= None 表示默认值为 None，如果没有提供该字段的值，则默认为 None。

3. name: str: 这是另一个属性声明，表示 Collection 类中的另一个字段。name 是一个必需的字符串类型，没有默认值，因此在创建 Collection 实例时必须提供 name 字段的值。

4. metadata: Dict = {}: 这是第三个属性声明，表示 Collection 类中的另一个字段。metadata 是一个字典类型，其中键值对的类型没有指定，因此可以接受任何类型的值。= {} 表示如果没有提供 metadata 字段的值，则默认为空字典 {}。

这个 Collection 类定义了一个数据模型，其中包含了 id、name 和 metadata 三个字段。id 字段是可选的，name 字段是必需的，而 metadata 字段是一个字典，可以包含任意附加信息。

4.字典默认值

 login_payload: dict
 values: dict = {}

在 Python 中，login_payload: dict 和 values: dict = {} 都是变量声明，它们之间的区别在于对于可选性和默认值的处理。

1. login_payload: dict： 这是一个变量声明，指定了变量名 login_payload 的类型为字典（dict）。但是，它并未提供默认值，因此在创建 login_payload 变量时必须显式地给它赋值，否则会抛出错误。

2. values: dict = {}： 这也是一个变量声明，指定了变量名 values 的类型为字典（dict）。不同之处在于它使用了默认值 = {}。这意味着如果在创建 values 变量时没有显式地给它赋值，它将默认为一个空字典 {}。这使得在代码中使用 values 变量时不必担心它是否已经被初始化。

因此，主要区别在于可选性和默认值的处理：

• login_payload 是一个必须显式赋值的变量，没有默认值。
• values 是一个可选的变量，如果没有提供值，则默认为一个空字典。

5.级联字典

user_name = session.login_payload['userInfo']['username']

这行代码是一个赋值语句，其目的是从一个名为 session 的对象中提取用户信息并赋值给变量 user_name。让我们逐步解释它：

1. session.login_payload['userInfo']['username']： 这是一个表达式，它首先通过 session 对象访问 login_payload 属性，然后从 login_payload 字典中获取键为 'userInfo' 的值。接着，再从 'userInfo' 对应的值中获取键为 'username' 的值。换句话说，它是从嵌套的字典结构中提取出用户名的操作。

2. user_name = ...： 这是一个赋值语句，它将上述表达式的结果赋值给变量 user_name。因此，user_name 变量最终将包含从 session.login_payload['userInfo']['username'] 表达式中提取出的用户名。

综上所述，这行代码的作用是从 session 对象的 login_payload 属性中提取出用户信息中的用户名，并将其赋值给变量 user_name。

6.日志

  logger.info("add_collection - login payload: %s", session.login_payload)

7.当前时间

time.time()

8.循环

循环遍历了一个可迭代对象 st 中的每个元素，并将每个元素依次赋值给变量 s。st 可以是一个列表、元组、字符串等可迭代对象。

具体解释如下：

for s in st:
    # 在此处对每个元素进行处理，可以使用变量 s

• for：这是 Python 中的一个关键字，用于循环遍历可迭代对象中的元素。
• s：这是循环中的变量名，它在每次迭代时会被赋值为可迭代对象 st 中的一个元素。
• in：这是另一个关键字，用于指示循环中要迭代的对象。
• st：这是要迭代的可迭代对象，它可以是列表、元组、字符串等。

在 for 循环内部，你可以对每个元素 s 进行任何操作，例如打印、处理等。例如：

st = [1, 2, 3, 4, 5]
for s in st:
    print(s)  # 输出列表中的每个元素

这将打印列表 st 中的每个元素，即数字 1 到 5。

另一个例子，如果 st 是一个字符串，你可以按照字符遍历它：

st = "hello"
for s in st:
    print(s)  # 逐个打印字符串中的字符

这将打印字符串 "hello" 中的每个字符，即 h、e、l、l、o。

9.init.py 文件的作用?

__init__.py 文件是 Python 包中的一个特殊文件，它的作用有几个方面：

1. 标识包目录：__init__.py 文件的存在将告诉 Python 解释器该目录是一个 Python 包。即使 __init__.py 文件为空，Python 也会将其识别为一个包。

2. 初始化包：__init__.py 文件可以包含初始化包的代码，例如设置包的默认配置、导入模块、定义变量或函数等。当导入一个包时，Python 会首先执行该包目录下的 __init__.py 文件。

3. 控制包的导入行为：在 __init__.py 文件中，你可以通过定义 __all__ 变量来控制包的导入行为。__all__ 是一个列表，包含了当前包中可以被导入的模块名。如果没有定义 __all__，默认情况下，只有不以下划线开头的模块会被导入。

4. 提供包级别的变量和函数：你可以在 __init__.py 文件中定义一些包级别的变量和函数，供包内的模块共享使用。

__init__.py 文件在 Python 包中具有多种作用，包括标识包目录、初始化包、控制导入行为和提供包级别的功能。

10.init 方法

在 Python 中，__init__ 方法是一个特殊的方法，用于在创建类的新实例时进行初始化操作。它是类的构造函数，在创建类的对象时自动调用。

下面是一个简单的示例说明了 __init__ 方法的用法：

class Person:
    def __init__(self, name, age):
        self.name = name
        self.age = age

# 创建一个 Person 类的实例
person1 = Person("Alice", 30)
print(person1.name)  # 输出："Alice"
print(person1.age)   # 输出：30

在这个示例中，Person 类有一个 __init__ 方法，该方法接受两个参数 name 和 age。在创建 Person 类的实例时，传递的参数会被用于初始化实例的 name 和 age 属性。

在 __init__ 方法中，第一个参数通常是 self，它表示实例本身。在方法内部，可以通过 self 来访问和操作实例的属性和方法。

__init__ 方法在创建实例时自动调用，因此你不需要手动调用它。当你创建类的实例时，Python 会自动调用 __init__ 方法，并将传递给实例的参数传递给 __init__ 方法中定义的参数。

总之，__init__ 方法用于在创建类的实例时执行一些初始化操作，以确保实例在创建后处于一个合适的状态。

11.设置实例

这段代码定义了一个 Python 类的 __init__ 方法，用于初始化该类的实例。让我解释一下这个 __init__ 方法的写法：

def __init__(
        self,
        type: str,  # docker or k8s
        msg: str = None,
) -> None:
    self.type = type
    self.msg = msg

这个 __init__ 方法接受两个参数 type 和 msg，以及一个隐式的 self 参数，用于引用类的实例。参数列表中的 self 是 Python 中约定的表示对象实例的引用，必须作为第一个参数出现在每个类方法的参数列表中。

• type: str：这是 __init__ 方法的第一个参数，它表示一个类型为字符串的参数，用于指定对象的类型。在代码中使用了类型注解，表明了参数的类型应该是字符串。

• msg: str = None：这是 __init__ 方法的第二个参数，它表示一个类型为字符串的参数，用于指定对象的消息。在代码中使用了类型注解，并且设置了默认值为 None，这意味着在创建对象时可以选择是否提供这个参数，如果不提供，则默认为 None。

在 __init__ 方法的函数体内，将传递给 __init__ 方法的参数值分别赋值给了对象实例的两个属性 self.type 和 self.msg。这样，在创建类的实例时，可以通过访问 self.type 和 self.msg 来获取或设置对象的类型和消息。

总的来说，这个 __init__ 方法用于初始化类的实例，并设置实例的类型和消息属性。

12.SQL 格式化

sql_count = f"""with t_user as (
  select uid as user_id, username from clAppUser where username = '{username}' limit 1
),
t_conv as (
  select * from clConversation inner join t_user on app_user_id = t_user.user_id where is_delete = false and mode = {mode}
)
select count(*) as n from Application as t_app inner join t_conv on t_conv.app_id = t_app.id where t_app.name like '%{search_text}%';"""

这段代码是一个 SQL 查询语句，使用了 Python 中的 f-string（格式化字符串）来插入变量的值。让我来解释一下它的写法和功能：

1. sql_count = f""" ... """： 这行代码定义了一个字符串变量 sql_count，使用了 f-string 格式化字符串的语法。f-string 允许在字符串中插入变量的值，并在大括号 {} 内使用变量名或表达式。

2. with t_user as (...) 和 with t_conv as (...)： 这是 SQL 中的 Common Table Expressions (CTE)（通用表达式），它们类似于临时表，允许在查询中创建命名的临时数据集合。t_user 和 t_conv 分别是两个命名的临时数据集，其中包含了从 clAppUser 表和 clConversation 表中检索的数据。

3. select count(*) as n from Application as t_app inner join t_conv on t_conv.app_id = t_app.id where t_app.name like '%{search_text}%';： 这是一个 SQL 查询语句，它使用了 t_conv 临时表作为基础数据集，并计算了符合条件的记录数。具体来说：

   • select count(*) as n 表示查询的结果是符合条件的记录数，并将结果别名为 n。
   • from Application as t_app 表示查询是从名为 Application 的表中进行的，并将其别名为 t_app。
   • inner join t_conv on t_conv.app_id = t_app.id 表示将 t_conv 临时表和 Application 表进行内连接，连接条件是 t_conv 表的 app_id 字段与 Application 表的 id 字段相匹配。
   • where t_app.name like '%{search_text}%' 表示对连接后的结果集进行条件过滤，只保留 Application 表中 name 字段中包含 search_text 变量值的记录。

总的来说，这段代码构建了一个复杂的 SQL 查询，使用了 CTE（通用表达式）和参数化查询，通过 Python 中的 f-string 插入变量值，以动态生成 SQL 查询语句。

13.路径参数问题

在提供的代码示例中，我们可以看到两个路由装饰器@KBRouter.post被应用于两个不同的函数。第一个函数get接收一个路径参数id，而第二个函数get则没有路径参数，但接收一个非路径参数info。两个函数的路由路径前缀都是/collection/docs，这可能导致一些混淆。

FastAPI 路径参数和非路径参数的前缀一致,当访问非路径参数的接口时,会走到路径参数的接口中

@staticmethod
@KBRouter.post("/collection/docs/{id}", description="路径参数接口",
               summary="路径参数接口")
async def get(id: str = None):
    return {
        "code": 0,
        "data": None,
        "msg": "成功"
    }

@staticmethod
@KBRouter.post("/collection/docs/add", description="非路径参数接口",
               summary="非路径参数接口")
async def get(info: str = None):
    return {
        "code": 0,
        "data": None,
        "msg": "成功",
    }

三.post 请求

1.客户端

def voc_list(app_id: str, words: List[str]):
    try:
        # 指定请求参数
        url: str = f"{api_base}/voc/list"
        data = {
            "app_id": app_id,
            "words": words
        }
        headers = {
            'accept': 'application/json',
            'Content-Type': 'application/json'
        }
        response = requests.post(url, data=json.dumps(data), headers=headers)
        # 检查响应状态码
        if response.status_code == 200:
            # 请求成功
            logger.info(f'voc list ok.:{response}')
            return response.text
        else:
            # 请求失败
            logger.error(f'voc list fail:{response}')
    except Exception as e:
        logger.error(f'voc list exception:{e}')

2.请求参数

class VocListRequest(BaseModel):
    app_id: str
    words: List[str]

3.服务端

@staticmethod
@AppRouter.post("/voc/list", summary="查询", description="查询")
async def voc_list(voc_list: VocListRequest):
   app_id = voc_list.app_id
   words = voc_list.words
   app = await AppManager().get_app_info(app_id)
   res = {}
   if app:
      vocab_table_ids = app.vocab_table_info
      if vocab_table_ids and len(vocab_table_ids) > 0:
            db = get_sqlalchemy_db
            with db.get_session() as session:
               vocabs = session.query(AlchemyVocab).filter(
                  AlchemyVocab.vocab_table_id.in_(vocab_table_ids)).filter(
                  AlchemyVocab.name.in_(words)).all()
               res = {vocab.name: vocab.desc for vocab in vocabs if vocab.desc is not None}
   return {
      "code": 0,
      "msg": "成功",
      "data": res
   }

四.基本介绍

1.dict 参数

from fastapi import FastAPI

app = FastAPI()

@app.post("/items/")
async def create_item(item: dict):
    return {"name": item["name"], "price": item["price"], "tax": item["price"] * 0.05}

2.发送 POST 请求

你可以使用 curl 命令行工具、Postman 或编写代码（如使用 Python 的 requests 库）来发送 POST 请求。以下是使用 curl 的示例：

curl -X 'POST' \
  'http://127.0.0.1:8000/items/' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "name": "Item1",
  "price": 10
}'

这将向 /items/ 路由发送一个 POST 请求，并包含 JSON 格式的数据。

3.对象参数

view 层

IndexRouter = APIRouter(prefix="/graph", tags=["图谱管理"])

class IndexRouterMap:

    @staticmethod
    @IndexRouter.post("/index_docs", summary="创建索引", description="创建索引")
    async def index_by_docs(kd: KnowledgeDocIndex):

        return {
            "code": 0,
            "data": None,
            "msg": "成功",
        }

model层

class KnowledgeDocIndex(BaseModel):
    kb_name: str
    schema_id: int = 2
    origNames: Optional[list[str]] = None
    run_id: Optional[str] = None

五.待整理

@staticmethod @TaskManageRouter.delete("/delete_by_kb_id", summary="根据 kb_id 删除图谱") async def delete_by_kb_id(kb_id: str):

要请求这个接口，你需要使用 HTTP DELETE 方法，并且将kb_id作为路径参数传递给这个 API。以下是一个使用 Python 中的aiohttp库来发送异步 HTTP DELETE 请求的示例代码：

import aiohttp
import asyncio

async def delete_kb_by_id(url, kb_id):
    async with aiohttp.ClientSession() as session:
        async with session.delete(f"{url}/delete_by_kb_id/{kb_id}") as response:
            if response.status == 200:
                data = await response.json()
                print("删除成功:", data)
            else:
                print("请求失败，状态码:", response.status)

# 假设你的API服务器地址是 http://api.example.com
api_url = "http://api.example.com"

# 要删除的kb_id
kb_id_to_delete = "some_kb_id"

# 运行异步任务
asyncio.run(delete_kb_by_id(api_url, kb_id_to_delete))

在这段代码中，delete_kb_by_id函数接受 API 的基础 URL 和要删除的kb_id作为参数。然后，它构造了正确的 URL，并通过aiohttp.ClientSession().delete方法发送 DELETE 请求。如果请求成功，它会打印出返回的数据。

请确保将api_url替换为你的实际 API 服务器地址，并且kb_id_to_delete替换为你想要删除的图谱的 ID。

如果你使用的是其他编程语言或 HTTP 客户端库，你需要调整上述代码以适应你的环境。例如，如果你使用的是 JavaScript 和 Fetch API，代码将会有所不同。