首页 智能体

# FastAPI

FastAPI 是一个用于构建 Web 应用的 Python 框架。

# 天生非阻塞

1 1

# 类型提示与验证

1

# 自带交互式文档

1

# 快速开始

from fastapi import FastAPI
# 创建一个 FastAPI 实例
app = FastAPI()
@app.get("/")
async def root():
    return {"message": "Hello World"}
@app.get("/hello/{name}")
async def say_hello(name: str):
    return {"message": f"Hello {name}"}
# 启动命令


uvicorn main:app --reload
uvicorn？ uvicorn  是一个高性能的 Python ASGI Web 服务器，专门用来运行 FastAPI / Starlette / Django ASGI 这类现代 Web 框架。


–reload？支持热部署。


# 结果

1
1
1

# Python 装饰器


在不改原函数代码的情况下，给函数外挂一层功能。


# Java 注解本身只是个标签


@GetMapping("/hello")
public String hello() {
    return "hi";
}
@GetMapping  本身不拦截，不执行，不改方法，真正干活的是 Spring 框架 + 反射 + AOP


# Python 装饰器是可执行的


def log(func):
    def wrapper(*args, **kwargs):
        print("before")
        return func(*args, **kwargs)
    return wrapper
@log
def hello():
    print("hi")
等价于：


hello = log(hello)
这里已经执行代码了


# 路由


@app.get("/hello/")
async def say_hello():
    return {"message": "Hello World 这是默输出"}
这是一个 get 方法，通过 @app.get  装饰。


# 接口的动态交互


路径参数写在 URL 路径里，用来标识具体资源


查询参数写在 URL  ?  后面，用来筛选 / 附加条件


请求体参数放在请求体中，用来提交复杂数据


# 路径参数


@app.get("/book/{id}")
async def getBook(id: int):
    return {
        "id": id,
        "title": f"FastAPI book-{id}"
    }
# 校验 —— 大小校验与长度校验


@app.get("/book/{id}")
async def getBook(id: int = Path(description="The ID of the book to get", gt=0, lt=1000)):
    return {
        "id": id,
        "title": f"FastAPI book-{id}"
    }
@app.get("/classify/{name}")
async def classify(name: str=Path(title="classify", min_length=3, max_length=10)):
    return {
        "name": name,
        "class": "FastAPI"
    }
# 查询参数


@app.get("/query_book")
async def query_book(
        classify: str = Query("默认分类", min_length=2, max_length=255),
        price: float = Query(10.0, gt=50.0, le=100.0)):
    return {
        "classify": classify,
        "price": price
    }
# 请求体参数


使用 pydantic。


from pydantic import BaseModel
class User(BaseModel):
    username: str
    password: str
@app.post("/register")
async def register(user: User):
    return user
class Book(BaseModel):
    title: str
    author: str
    publisher: str
    price: float
@app.post("/book/add")
async def add_book(book: Book):
    return book
请求体参数的约束：


class User(BaseModel):
    username: str = Field(default="默认用户名", min_length=2, max_length=10, description="用户名")
    password: str = Field(default="默认密码", min_length=6, max_length=20, description="密码")
@app.post("/register")
async def register(user: User):
    return user
# 响应类型


通过 response_class 指定

1

# JSON


这个你返回一个字典对象的时候，FastAPI 就自动帮你做好转换了。


# HTML / 纯文本


@app.get("/html",response_class=HTMLResponse)
async def html():
    return """
    <html>
        <head>
            <title>FastAPI</title>
        </head>
        <body>
            <h1>Hello World</h1>
        </body>
    </html>
    """
# File 文件


@app.get("/file")
async def file():
    path = "./files/dd.txt"
    return FileResponse(path)
# 自定义响应类型


class R(BaseModel):
    code: int
    msg: str
    data: dict = Field(default={}, description="返回数据")
@app.get("/api", response_model=R)
async def api():
    return {
        "code": 200,
        "msg": "success",
        "data": {
            "name": "KarryLiu",
            "age": 18
        },
    }
当你缺失数据时会报错，当你数据超出约定时会自动过滤掉。


# 异常处理


@app.get("/api/news", response_model=R)
async def api_news(id: int = Query(1, gt=0, lt=1000)):
    news_list = [1, 2, 3]
    if id not in news_list:
        raise HTTPException(status_code=404, detail="新闻不存在")
    return {
        "code": 200,
        "msg": "success",
        "data": {
            "id": id,
            "title": f"新闻标题-{id}",
            "content": f"新闻内容-{id}"
        },
    }
# 中间件


中间件给每一个请求提供统一的处理逻辑。使用 @app.middleware("http")  来定义。多个中间件是自底向上执行的


@app.middleware("http")
async def middleware1(request, call_next):
    print("中间件1开始执行")
    response = await call_next(request)
    print("中间件1结束执行")
    return response
@app.middleware("http")
async def middleware2(request, call_next):
    print("中间件2开始执行")
    response = await call_next(request)
    print("中间件2结束执行")
    return response
@app.get("/api/mid")
async def api_mid():
    return {
        "code": 200,
        "msg": "success",
        "data": {
            "name": "KarryLiu",
            "age": 18
        },
    }
1

居然是自底向上执行的


# 依赖注入

1
1

async def common_parameters(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}
@app.get("/api/news_list")
async def get_news_list(commons=Depends(common_parameters)):
    return {"msg": "news_list"}
@app.get("api/user_list")
async def get_user_list(commons=Depends(common_parameters)):
    return {"msg": "user_list"}
common_parameters  里  return {"skip": skip, "limit": limit}  返回值不会直接返回给客户端， 而是 作为参数值，传给  get_news_list  里的  commons  变量。我们可以打印一下：


@app.get("/api/news_list")
async def get_news_list(commons=Depends(common_parameters)):
    print(commons)
    print(commons["skip"])
    print(commons["limit"])
    return {"msg": "news_list
1

# ORM 工具 ——SQLAlchemy


# 安装 sqlalchemy [asyncio] 和 aiomysql


pip install sqlalchemy[asyncio] aiomysql
# 自动建表


ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/fastapi_first?charsetutf-8"
engine = create_async_engine(ASYNC_DATABASE_URL, echo=True, pool_size=10, max_overflow=20, pool_recycle=3600,
                             pool_pre_ping=True, pool_use_lifo=True, pool_timeout=30, future=True, )
class Base(DeclarativeBase):
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), comment="创建时间")
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), comment="更新时间")
class Book(Base):
    __tablename__ = "book"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True, comment="ID")
    title: Mapped[str] = mapped_column(String(100), comment="标题")
    author: Mapped[str] = mapped_column(String(50), comment="作者")
    publisher: Mapped[str] = mapped_column(String(100), comment="出版社")
    price: Mapped[float] = mapped_column(Float, comment="价格")
class User(Base):
    __tablename__ = "user"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True, comment="ID")
    username: Mapped[str] = mapped_column(String(50), comment="用户名")
    password: Mapped[str] = mapped_column(String(50), comment="密码")
async def create_table():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
@asynccontextmanager
async def lifespan(apps: FastAPI):
    await create_table()
    yield
    await engine.dispose()
app = FastAPI(lifespan=lifespan)
1

# ORM 依赖注入与数据查询


AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)
async def get_database():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception as e:
            await session.rollback()
            raise
        finally:
            await session.close()
@app.get("/book/books")
async def get_books(session: AsyncSession = Depends(get_database)):
    query = select(Book)
    result = await session.execute(query)
    books = result.scalars().all()
    print(books)
    return books
1

# 查询


# 比较判断


==；>；<；>=；<=


@app.get("/book/book/{id}")
async def get_book(id: int, session: AsyncSession = Depends(get_database)):
    query = select(Book).where(Book.id == id)
    result = await session.execute(query)
    book = result.scalars().first()
    return book
@app.get("/book/book/{id}")
async def get_book(id: int, session: AsyncSession = Depends(get_database)):
    query = select(Book).where(Book.id == id)
    result = await session.execute(query)
    book = result.scalars().one_or_none()
    return book
# 模糊查询


@app.get("/book/like")
async def get_like_book(title: str, session: AsyncSession = Depends(get_database)):
    query = select(Book).where(Book.title.like(f"%{title}%"))
    result = await session.execute(query)
    book = result.scalars().all()
    return book
1

# 取首条数据


@app.get("/book/first")
async def get_first_book(session: AsyncSession = Depends(get_database)):
    query = select(Book)
    result = await session.execute(query)
    book = result.scalars().first()
    return book
# 多条件


@app.get("/book/like_start")
async def get_like_start_book(title: str, session: AsyncSession = Depends(get_database)):
    query = select(Book).where(Book.title.like(f"{title}%")).where(Book.id == 2)
    result = await session.execute(query)
    book = result.scalars().all()
    return book
# 聚合查询


func——count / avg / max / min / sum


@app.get("/book/get_avg_price")
async def get_avg_price(session: AsyncSession = Depends(get_database)):
    query = select(func.avg(Book.price))
    result = await session.execute(query)
    avg_price = result.scalars().one()
    return avg_price
1
1

      

          
          
          
      

  


   


    

  
    
      
    
    更新于
    
  












      


  
  
请我喝[茶]~(￣▽￣)~*

  

      
      

        KarryLiu 微信支付
        
微信支付

      

      
      

        KarryLiu 支付宝
        
支付宝