首页 星云 工具 资源 星选 资讯 热门工具
:

PDF转图片 完全免费 小红书视频下载 无水印 抖音视频下载 无水印 数字星空

使用FastAPI来开发项目,项目的目录结构如何规划的一些参考和基类封装的一些处理

编程知识
2024年08月28日 08:52

使用FastAPI开发项目时,良好的目录结构可以帮助你更好地组织代码,提高可维护性和扩展性。同样,对基类的封装,也可以进一步减少开发代码,提供便利,并减少出错的几率。

下面是一个推荐的目录结构示例:

my_fastapi_project/
├── app/
│   ├── __init__.py
│   ├── main.py            # 入口文件
│   ├── core/
│   │   ├── __init__.py
│   │   ├── config.py      # 配置文件
│   │   ├── security.py    # 安全相关
│   │   └── ...            # 其他核心功能
│   ├── api/
│   │   ├── __init__.py
│   │   ├── v1/
│   │   │   ├── __init__.py
│   │   │   ├── endpoints/
│   │   │   │   ├── __init__.py
│   │   │   │   ├── users.py     # 用户相关接口
│   │   │   │   ├── items.py     # 其他接口
│   │   │   │   └── ...
│   │   │   └── ...              # 其他版本的API
│   ├── models/
│   │   ├── __init__.py
│   │   ├── user.py         # 用户模型
│   │   ├── item.py         # 其他模型
│   │   └── ...
│   ├── schemas/
│   │   ├── __init__.py
│   │   ├── user.py         # 用户数据模型
│   │   ├── item.py         # 其他数据模型
│   │   └── ...
│   ├── crud/
│   │   ├── __init__.py
│   │   ├── user.py         # 用户CRUD操作
│   │   ├── item.py         # 其他CRUD操作
│   │   └── ...
│   ├── db/
│   │   ├── __init__.py
│   │   ├── base.py         # 数据库基础设置
│   │   ├── session.py      # 数据库会话
│   │   └── ...
│   ├── tests/
│   │   ├── __init__.py
│   │   ├── test_main.py    # 测试主文件
│   │   ├── test_users.py   # 用户相关测试
│   │   └── ...
│   └── utils/
│       ├── __init__.py
│       ├── utils.py        # 工具函数
│       └── ...
├── .env                    # 环境变量文件
├── alembic/                # 数据库迁移工具目录
│   ├── env.py
│   ├── script.py.mako
│   └── versions/
│       └── ...
├── alembic.ini             # Alembic 配置文件
├── requirements.txt        # 项目依赖
├── Dockerfile              # Docker 配置文件
└── README.md               # 项目说明文件

目录结构说明:

  • app/: 项目的主目录,包含所有应用相关代码。
    • main.py: 项目的入口文件,启动FastAPI应用。
    • core/: 核心功能,如配置、安全等。
    • api/: API路由和视图,分版本管理。
    • models/: 数据库模型。
    • schemas/: 数据模型,用于请求和响应的验证。
    • crud/: 数据库操作(CRUD:创建、读取、更新、删除)。
    • db/: 数据库相关设置和会话管理。
    • tests/: 测试代码。
    • utils/: 工具函数和公用模块。
  • .env: 环境变量文件,用于存储敏感信息,如数据库连接字符串。
  • alembic/: 数据库迁移工具Alembic的配置目录。
  • requirements.txt: 项目依赖列表。
  • Dockerfile: Docker配置文件,用于容器化部署。
  • README.md: 项目说明文件。

这个结构可以根据项目需求进行调整,但保持清晰和模块化是良好的实践。

 

python项目总的__init__.py,有意义吗

在Python项目中,__init__.py 文件的主要作用是将目录标识为一个Python包。它使得目录中的模块可以被导入和使用。在一些情况下,__init__.py 可以不仅仅是一个空文件,还可以包含一些初始化代码。

__init__.py 的意义:

  1. 将目录标识为包:

    • 任何包含 __init__.py 的目录都会被Python解释器认为是一个包,这样你就可以使用包导入语法,如 import mypackage.module
  2. 初始化代码:

    • 可以在 __init__.py 中包含一些初始化代码,如导入包内的子模块、设置包级别的变量或函数、配置日志记录等。例如:
# mypackage/__init__.py
from .submodule1 import func1
from .submodule2 import func2

__all__ = ["func1", "func2"]

        3.简化导入

    •  通过在 __init__.py 中导入子模块,可以简化包的导入路径,使得用户可以直接从包中导入函数或类,而不必知道具体的模块结构。
# mypackage/__init__.py
from .submodule import MyClass

# Now you can do
from mypackage import MyClass

对于Python 3.3及以上版本,__init__.py 文件不是强制性的,即使没有 __init__.py 文件,Python解释器也可以识别包。然而,添加 __init__.py 文件仍然是一个良好的习惯,可以避免某些情况下的意外行为,并且明确表示该目录是一个包。

 

2、Fast API项目的开发处理过程

在FastAPI项目中,CRUD操作通常在一个专门的 crud 模块中实现。这个模块会调用SQLAlchemy模型对象来进行数据库操作。

1. 定义模型 (models/user.py)

from sqlalchemy import Column, Integer, String
from app.db.base_class import Base

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    full_name = Column(String, index=True)

2. 创建数据库会话 (db/session.py)

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./test.db"  # 使用SQLite数据库作为示例

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

3. 定义CRUD操作 (crud/user.py)

from sqlalchemy.orm import Session
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate

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

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

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

def create_user(db: Session, user: UserCreate):
    db_user = User(
        email=user.email,
        hashed_password=user.hashed_password,  # 在实际应用中应该对密码进行哈希处理
        full_name=user.full_name
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def update_user(db: Session, user_id: int, user: UserUpdate):
    db_user = get_user(db, user_id)
    if db_user:
        db_user.email = user.email
        db_user.full_name = user.full_name
        db.commit()
        db.refresh(db_user)
    return db_user

def delete_user(db: Session, user_id: int):
    db_user = get_user(db, user_id)
    if db_user:
        db.delete(db_user)
        db.commit()
    return db_user

4. 定义数据模型 (schemas/user.py)

from pydantic import BaseModel

class UserBase(BaseModel):
    email: str
    full_name: str = None

class UserCreate(UserBase):
    hashed_password: str

class UserUpdate(UserBase):
    pass

class User(UserBase):
    id: int

    class Config:
        orm_mode = True

5. 在API端点中使用CRUD操作 (api/v1/endpoints/users.py)

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app import crud, models, schemas
from app.db.session import SessionLocal

router = APIRouter()

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

@router.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)

@router.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

@router.put("/users/{user_id}", response_model=schemas.User)
def update_user(user_id: int, user: schemas.UserUpdate, db: Session = Depends(get_db)):
    db_user = crud.update_user(db=db, user_id=user_id, user=user)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

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

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

6. 注册路由 (main.py)

from fastapi import FastAPI
from app.api.v1.endpoints import users

app = FastAPI()

app.include_router(users.router, prefix="/api/v1", tags=["users"])

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

7. 初始化数据库 (db/base.py)

from app.db.session import engine
from app.models import user

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

8. 运行应用

在项目根目录下运行:

uvicorn app.main:app --reload

这样,你的CRUD层就可以调用模型对象来进行数据库操作了。上述代码展示了如何定义模型、数据库会话、CRUD操作、数据模型和API端点,并将它们结合在一起,实现一个简单的用户管理系统。

 

3、实际FastAPI项目对基类的封装

可以通过创建一个通用的CRUD基类来封装常规的CRUD操作,然后让特定的CRUD类继承这个基类。这样可以减少重复代码,提高代码的可维护性和可复用性。下面是一个实现示例。

1、创建通用CRUD基类 (crud/base.py)

rom typing import Generic, Type, TypeVar, Optional, List
from pydantic import BaseModel
from sqlalchemy.orm import Session
from app.db.base_class import Base

ModelType = TypeVar("ModelType", bound=Base)
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)

class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
    def __init__(self, model: Type[ModelType]):
        self.model = model

    def get(self, db: Session, id: int) -> Optional[ModelType]:
        return db.query(self.model).filter(self.model.id == id).first()

    def get_multi(self, db: Session, skip: int = 0, limit: int = 100) -> List[ModelType]:
        return db.query(self.model).offset(skip).limit(limit).all()

    def create(self, db: Session, obj_in: CreateSchemaType) -> ModelType:
        obj_in_data = obj_in.dict()
        db_obj = self.model(**obj_in_data)  # type: ignore
        db.add(db_obj)
        db.commit()
        db.refresh(db_obj)
        return db_obj

    def update(self, db: Session, db_obj: ModelType, obj_in: UpdateSchemaType) -> ModelType:
        obj_data = db_obj.dict()
        update_data = obj_in.dict(skip_defaults=True)
        for field in obj_data:
            if field in update_data:
                setattr(db_obj, field, update_data[field])
        db.commit()
        db.refresh(db_obj)
        return db_obj

    def remove(self, db: Session, id: int) -> ModelType:
        obj = db.query(self.model).get(id)
        db.delete(obj)
        db.commit()
        return obj

2、定义用户CRUD操作 (crud/user.py)

from typing import Any
from sqlalchemy.orm import Session
from app.crud.base import CRUDBase
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate

class CRUDUser(CRUDBase[User, UserCreate, UserUpdate]):
    def get_by_email(self, db: Session, email: str) -> Any:
        return db.query(self.model).filter(self.model.email == email).first()

user = CRUDUser(User)

3、定义数据模型 (schemas/user.py)

from pydantic import BaseModel

class UserBase(BaseModel):
    email: str
    full_name: str = None

class UserCreate(UserBase):
    hashed_password: str

class UserUpdate(UserBase):
    pass

class User(UserBase):
    id: int

    class Config:
        orm_mode = True

4、在API端点中使用CRUD操作 (api/v1/endpoints/users.py)

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List
from app import crud, schemas
from app.db.session import SessionLocal
from app.models.user import User

router = APIRouter()

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

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

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

@router.put("/users/{user_id}", response_model=schemas.User)
def update_user(user_id: int, user: schemas.UserUpdate, db: Session = Depends(get_db)):
    db_user = crud.user.get(db=db, id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return crud.user.update(db=db, db_obj=db_user, obj_in=user)

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

@router.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    users = crud.user.get_multi(db, skip=skip, limit=limit)
    return users

其他的就是类似前面的做法了。

通过这种方式,你可以在通用的CRUD基类中封装常规的CRUD操作,而特定的CRUD类(如 CRUDUser)只需要继承这个基类并添加特定的操作方法。这样不仅减少了重复代码,也提高了代码的可维护性和可复用性。

如果你希望可以通过定义一个通用的API基类来封装常规的CRUD操作方法,然后在具体的端点文件中继承这个基类。这样可以进一步减少重复代码,提高代码的可维护性和可复用性。

 创建通用API基类 (api/deps.py)

from typing import Type, TypeVar, Generic, List
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from pydantic import BaseModel
from app.crud.base import CRUDBase
from app.db.session import SessionLocal

ModelType = TypeVar("ModelType", bound=BaseModel)
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)

class CRUDRouter(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
    def __init__(self, crud: CRUDBase[ModelType, CreateSchemaType, UpdateSchemaType]):
        self.crud = crud
        self.router = APIRouter()

        self.router.post("/", response_model=ModelType)(self.create_item)
        self.router.get("/{item_id}", response_model=ModelType)(self.read_item)
        self.router.put("/{item_id}", response_model=ModelType)(self.update_item)
        self.router.delete("/{item_id}", response_model=ModelType)(self.delete_item)
        self.router.get("/", response_model=List[ModelType])(self.read_items)

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

    async def create_item(self, item_in: CreateSchemaType, db: Session = Depends(self.get_db)):
        db_item = self.crud.create(db=db, obj_in=item_in)
        return db_item

    async def read_item(self, item_id: int, db: Session = Depends(self.get_db)):
        db_item = self.crud.get(db=db, id=item_id)
        if not db_item:
            raise HTTPException(status_code=404, detail="Item not found")
        return db_item

    async def update_item(self, item_id: int, item_in: UpdateSchemaType, db: Session = Depends(self.get_db)):
        db_item = self.crud.get(db=db, id=item_id)
        if not db_item:
            raise HTTPException(status_code=404, detail="Item not found")
        return self.crud.update(db=db, db_obj=db_item, obj_in=item_in)

    async def delete_item(self, item_id: int, db: Session = Depends(self.get_db)):
        db_item = self.crud.get(db=db, id=item_id)
        if not db_item:
            raise HTTPException(status_code=404, detail="Item not found")
        return self.crud.remove(db=db, id=item_id)

    async def read_items(self, skip: int = 0, limit: int = 10, db: Session = Depends(self.get_db)):
        items = self.crud.get_multi(db=db, skip=skip, limit=limit)
        return items

使用通用API基类定义用户端点(api/v1/endpoints/users.py)

from fastapi import APIRouter
from app.crud.user import user as user_crud
from app.schemas.user import User, UserCreate, UserUpdate
from app.api.deps import CRUDRouter

user_router = CRUDRouter[User, UserCreate, UserUpdate](user_crud)
router = user_router.router

注册路由 (main.py)

rom fastapi import FastAPI
from app.api.v1.endpoints import users

app = FastAPI()

app.include_router(users.router, prefix="/api/v1/users", tags=["users"])

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

通过这种方式,你可以在 CRUDRouter 基类中封装常规的CRUD操作方法,然后在具体的端点文件中继承这个基类并传递相应的CRUD对象。这样可以进一步减少重复代码,提高代码的可维护性和可复用性。

 

4、SQLAlchemy模型的基类定义

app.db.base_class 通常是用于定义SQLAlchemy模型基类的文件。在这个文件中,我们会定义一个基本的Base类,这个类是所有SQLAlchemy模型的基类。下面是一个实现示例:

定义 Base 类 (db/base_class.py)

from sqlalchemy.ext.declarative import as_declarative, declared_attr

@as_declarative()
class Base:
    id: int
    __name__: str

    @declared_attr
    def __tablename__(cls) -> str:
        return cls.__name__.lower()

详细解释

  1. @as_declarative(): 这是SQLAlchemy提供的一个装饰器,它会将类装饰为一个声明性基类。所有继承自这个类的子类都会自动成为声明性类。

  2. id: int: 这是一个类型注释,表示每个模型类都会有一个 id 属性。具体的字段定义(例如 Column(Integer, primary_key=True))会在每个具体的模型类中定义。

  3. __name__: str: 这是另一个类型注释,表示每个模型类都会有一个 __name__ 属性。

  4. @declared_attr: 这是SQLAlchemy提供的一个装饰器,允许我们为声明性基类定义一些通用的属性。在这个例子中,它用于自动生成 __tablename__ 属性。这个属性的值是模型类的名称的小写形式。

这样定义的 Base 类可以作为所有SQLAlchemy模型的基类,简化模型的定义。

完整示例项目结构:为了更好地理解,这里展示一个包含 Base 类定义的完整项目结构:

.
├── app
│   ├── __init__.py
│   ├── api
│   │   ├── __init__.py
│   │   └── v1
│   │       ├── __init__.py
│   │       └── endpoints
│   │           ├── __init__.py
│   │           └── users.py
│   ├── crud
│   │   ├── __init__.py
│   │   ├── base.py
│   │   └── user.py
│   ├── db
│   │   ├── __init__.py
│   │   ├── base.py
│   │   ├── base_class.py
│   │   └── session.py
│   ├── models
│   │   ├── __init__.py
│   │   └── user.py
│   ├── schemas
│   │   ├── __init__.py
│   │   └── user.py
│   └── main.py

models/user.py 类文件如下定义

from sqlalchemy import Column, Integer, String
from app.db.base_class import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    full_name = Column(String, index=True)

通过这种结构和定义,您可以创建一个简洁、可扩展的FastAPI项目,能够快速定义新的数据库模型并生成相应的CRUD操作和API端点。

 

From:https://www.cnblogs.com/wuhuacong/p/18380808
本文地址: http://www.shuzixingkong.net/article/1509
0评论
提交 加载更多评论
其他文章 推荐一款开源一站式SQL审核查询平台!功能强大、安全可靠!
1、前言 在当今这个数据驱动的时代,数据库作为企业核心信息资产的载体,其重要性不言而喻。随着企业业务规模的不断扩大,数据库的数量和种类也日益增多,这对数据库的管理与运维工作提出了前所未有的挑战。在这样的背景下,一款高效、易用的数据库管理工具显得尤为重要。Archery,作为一款开源的数据库管理平台,
推荐一款开源一站式SQL审核查询平台!功能强大、安全可靠! 推荐一款开源一站式SQL审核查询平台!功能强大、安全可靠! 推荐一款开源一站式SQL审核查询平台!功能强大、安全可靠!
巴特沃斯LPF设计(硬件电路实现)
高阶 (2n) VSVC单位增益巴特沃斯低通滤波器设计,可分解为 n 个二阶低通;本文通过对二阶巴特沃斯低通的硬件电路设计,后期多组组合优化,构建高阶滤波器,优化滤波器的低通特性和稳定性。
巴特沃斯LPF设计(硬件电路实现) 巴特沃斯LPF设计(硬件电路实现) 巴特沃斯LPF设计(硬件电路实现)
《HelloGitHub》第 101 期
每月 28 号更新的开源月刊,这里有实战项目、入门教程、黑科技、开源书籍、大厂开源项目等,涵盖多种编程语言 Python、Java、Go、C/C++、Swift...让你在短时间内感受到开源的魅力,对编程产生兴趣!
《HelloGitHub》第 101 期 《HelloGitHub》第 101 期 《HelloGitHub》第 101 期
Kafka Topic 中明明有可拉取的消息,为什么 poll 不到
开心一刻 今天小学女同学给我发消息她:你现在是毕业了吗我:嗯,今年刚毕业她给我发了一张照片,怀里抱着一只大橘猫她:我的眯眯长这么大了,好看吗我:你把猫挪开点,它挡住了,我看不到她:你是 sb 吗,滚我解释道:你说的是猫呀可消息刚发出,就出现了红色感叹号,并提示:消息已发出,但被对方拒收了 kafka
Kafka Topic 中明明有可拉取的消息,为什么 poll 不到 Kafka Topic 中明明有可拉取的消息,为什么 poll 不到 Kafka Topic 中明明有可拉取的消息,为什么 poll 不到
CSEC:香港城市大学提出SOTA曝光矫正算法 | CVPR 2024
在光照条件不佳下捕获的图像可能同时包含过曝和欠曝。目前的方法主要集中在调整图像亮度上,这可能会加剧欠曝区域的色调失真,并且无法恢复过曝区域的准确颜色。论文提出通过学习估计和校正这种色调偏移,来增强既有过曝又有欠曝的图像。先通过基于UNet的网络推导输入图像的增亮和变暗版本的色彩特征图,然后使用伪正常
CSEC:香港城市大学提出SOTA曝光矫正算法 | CVPR 2024 CSEC:香港城市大学提出SOTA曝光矫正算法 | CVPR 2024 CSEC:香港城市大学提出SOTA曝光矫正算法 | CVPR 2024
折腾 Quickwit,Rust 编写的分布式搜索引擎 - 从不同的来源摄取数据
摄取 API 在这节教程中,我们将介绍如何使用 Ingest API 向 Quickwit 发送数据。 要跟随这节教程,您需要有一个本地的 Quickwit 实例正在运行。 https://quickwit.io/docs/get-started/installation 要启动它,请在终端中运行
折腾 Quickwit,Rust 编写的分布式搜索引擎 - 从不同的来源摄取数据
实用接地气的 .NET 微服务框架
前言 微服务架构已经成为搭建高效、可扩展系统的关键技术之一,然而,现有许多微服务框架往往过于复杂,使得我们普通开发者难以快速上手并体验到微服务带了的便利。为了解决这一问题,于是作者精心打造了一款最接地气的 .NET 微服务框架,帮助我们轻松构建和管理微服务应用。 本框架不仅支持 Consul 服务注
实用接地气的 .NET 微服务框架 实用接地气的 .NET 微服务框架 实用接地气的 .NET 微服务框架
OpenTelemetry 实战:从零实现应用指标监控
前言 在上一篇文章:OpenTelemetry 实战:从零实现分布式链路追踪讲解了链路相关的实战,本次我们继续跟进如何使用 OpenTelemetry 集成 metrics 监控。 建议对指标监控不太熟的朋友可以先查看这篇前菜文章:从 Prometheus 到 OpenTelemetry:指标监控的
OpenTelemetry 实战:从零实现应用指标监控 OpenTelemetry 实战:从零实现应用指标监控 OpenTelemetry 实战:从零实现应用指标监控