牡丹江市网站建设_网站建设公司_Banner设计_seo优化

学习Python FastAPI

学习资料：
视频地址 1.01-【异步开发】异步开发相关概念_哔哩哔哩_bilibili  

如有帮助还请给up主 关注加三连

《零基础玩转FastAPI资料》
1. 代码：
链接: https://pan.baidu.com/s/1XrX83mPWIOcMYdq5wX-tSg?pwd=zb59 提取码: zb59

2. 课件：
https://www.yuque.com/hynever/wms8gi?# 《知了传课-FastAPI》 密码：puvz

3.mysql 安装教程
2024 年 MySQL 8.0 安装 配置 教程 最简易（保姆级）_mysql安装-CSDN博客

4.NavCat 16版本安装包 在这个链接里面下载
MySQL & NaviCat 安装及配置教程（Windows）【安装】_navicat下载安装和配置-CSDN博客

破解教程看这个：
Navicat Premium 16 安装并破解激活图文教程（亲测可行） - 个人博客

后续浏览其他的资料也会更新在这里 conda activate python3.9
python环境配置：【轮椅级教程】pytorch环境配置_哔哩哔哩_bilibili  很详细

创建python中第一个协程

import asyncio
# import time
async def name():print('我是后裔')# 协程必须要等待， 需要在前面加上awaitawait asyncio.sleep(1)# 在携程代码中对于那些会发生阻塞的 i/o代码，一定不能使用同步的 # 例如 time.sleep()print('我是嫦娥')# 这样执行 name() 并不会执行 直接执行 需要丢在 事件循环中 asyncio.run()
# name()
asyncio.run(name())

 计算时间装饰器

from functools import wraps
import time
def time_this(func):@wraps(func)async def wrapper(*args, **kwarges):start = time.time()try:return await func(*args, **kwarges)finally:print("{0} 耗时 {1:.2f} 秒".format(func.__name__, time.time() - start))return wrapper
print("functools 模块可用！")

使用

import asyncio
from utils import time_this
# import time
@time_this
async def name():
    print('我是后裔')
    # 协程必须要等待， 需要在前面加上await
    await asyncio.sleep(1)
    await asyncio.sleep(1)
    await asyncio.sleep(1)
    await asyncio.sleep(1)
    # 在携程代码中对于那些会发生阻塞的 i/o代码，一定不能使用同步的 否则程序就会阻塞在代码这里失去并发性
    # 例如 time.sleep()
    print('我是嫦娥')


# 这样执行 name() 并不会执行 直接执行 需要丢在 事件循环中 asyncio.run()
# name()
asyncio.run(name())

 

放在 执行函数的上面@符号相当于装饰器的标识

 异步执行python

import asyncio
from  utils import time_thisasync def green(name, delay):await asyncio.sleep(delay)return f'hello {name}'# 这是同步运行
# @time_this
# async def main():
#     a = await green('world', 1)
#     print(a)
#     a1 = await green('world2', 2)
#     print(a1)# asyncio.run(main())# 用创建任务的方式 来并发运行@time_this
async def main():# 并发运行必须要将协程包装成 TASK 对象 才能够并发执行task =  asyncio.create_task(green('world', 1))task1 =  asyncio.create_task(green('world111', 2))# 对创建好的任务进行awaita = await taska1 = await task1print(a, a1)
# asyncio.run(main())# 错误写法
#  err = await asyncio.create_task(green('world111', 2))
# 核心要点
# 1.asyncio.create_task() 的作用是将协程加入事件循环并立即开始执行
# 2.await 是用来获取任务结果的操作，会阻塞直到任务完成
# 3.要实现真正的并发，应该先创建所有任务，然后才逐一等待结果
# 一句话总结
# 先创建所有任务再依次等待结果可以实现并发执行（总耗时约2秒），而创建任务后立即await会变成串行执行（总耗时约3秒）。async def green_err(name, delay):await asyncio.sleep(delay)if name == 'xx':raise Exception('执行错误!')return f'hello {name}'
# 2. 用task任务组方式并发运行
async def mainTask():try:async with asyncio.TaskGroup() as tg:task = tg.create_task(green_err('world111', 3))task1 = tg.create_task(green('world222', 2))#   done: 返回该协程是否完成(被取消也算是完成)#   cancelled: 返回该协程是否被取消print(task.cancelled(), task1.done(), '查看导出模块')# 上面代码执行过后，有可能是两种情况# 1.所有任务执行完成# 2.其中有任务报错或者异常 导致后面的任务被取消,从而提前退出协程的运行except Exception as e:print(e, '任务组报错信息')# asyncio.run(mainTask())# 3.用gahter方式并发运行
async def mainGather():try:# gather()执行结束之后会按照入队形式 存放在res 中res = await asyncio.gather(green_err('xx', 1),green('yy', 2),green('zz', 3), return_exceptions=True)print(res)except Exception as e:print(e, 'gather错误')# asyncio.run(mainGather())# 4.使用as_completed方式并发运行
async def mainAsCompleted():try:res =  asyncio.as_completed([green('xx', 4),green('yy', 1),green('zz', 2), ], timeout=2)for i in res:iRes = await iprint(iRes)except Exception as e:print(e, 'as_completed错误')
asyncio.run(mainAsCompleted())

第一个fastApi

import typing as t
from fastapi import FastAPIapp =  FastAPI()@app.get("/")async def root():return {"message": "Hello World"}@app.get('/item/{item_id}')
async def method_name(item_id: int, q: t.Optional[str] = '参数是空的'):return {"item_id": item_id, "q": q}运行命令 ：uvicorn fastApi:app --reload --port 5678

fastApi 就是文件夹的名称 去掉了.py

下载了 fastapi 

uvicorn  来帮我们完成本地 链接建立  

http://127.0.0.1:5678 就是咱们的端口号了

类型约束

from typing import List,Dict,Tuple,Set,Union, Optional, Any# 1. list: 定义一个整数列表，包含三个元素1, 2, 3
lists: list[int] = [1, 2, 3]# 2. dict: 定义一个字典，键为字符串类型，值为整数类型
# 字典包含两个键值对：'name': 1 和 'age': 10
dicts: dict[str, int] = {'name': 1, 'age': 10}# 3. tuple: 定义一个元组，第一个元素为整数，第二个元素为字符串
# 元组包含值：1 和 '效力'
tiple: tuple[int, str] = (1, '效力')# 4. set: 定义一个整数集合，包含三个不重复的元素：1, 2, 3
sets: set[int] = {1, 2, 3}# 5. union: 定义一个联合类型变量，可以是字符串或整数类型
unions: Union[str, int] = 1# 使用管道符(|)定义联合类型，可以是字符串、整数或None类型
strint: str | int | None = None# 6. optional: 定义一个可选类型变量，可以是整数或None
# Optional[int]等价于Union[int, None]
optionals: Optional[int] = None# 7. any: 定义一个任意类型变量，默认情况下所有类型都可以赋值
# any表示不进行类型检查
anys: Any  = 1
# 8.类型别名 
listStr = List[str]listName: listStr = ['小王', '小李', '小张']def main(age: int, name: str) -> str:return f'{name} is {age} years old'print(main(18,  '小李'))

 接受参数

from fastapi import FastAPI, Path, Query, Depends, Body
from pydantic import BaseModel
# from typing import Dict, Any
app = FastAPI()
# 1、get请求
@app.get("/")
async def root():return {"message": "Hello World"}# 路由上的路径参数
@app.get('/item/{nameItem}')
async def names(nameItem: str = Path(max_length=10, description="最长10个字符")) -> str:return f"{nameItem}"# 路由上的查询参数
@app.get('book')
async def book(page: int = None, name: str = None) -> str:return f"{name}"# 路由上的查询参数
@app.get('lists')
async def book(page: int = Query(default= 1, description='名称'), name: str = None) -> str:return f"{name}"# 2.post请求
#   2.1 get 函数依赖 实现
def depend(qa: str, limit: int, page: int):print("qa:", qa, "limit:", limit, "page:", page)return {"qa": qa, "limit": limit, "page": page}class loginItme(BaseModel):qa: strlimit: intpage: int
@app.get('/login')
# post Depends 不生效 get 生效
async def login(body_item: dict = Depends(depend)       # 处理请求体参数
):print("Body item:", body_item)return {"message": "success"}
# 2.2 post 请求体参数 依赖项实现def dependpost(qa: str = Body(...), limit: int = Body(...), page: int = Body(...)):print("qa:", qa, "limit:", limit, "page:", page)return {"qa": qa, "limit": limit, "page": page}class loginItme(BaseModel):qa: strlimit: intpage: int
@app.post('/logins')
# post Depends 不生效 get 生效
async def login(body_item: dict = Depends(dependpost)       # 处理请求体参数
):print("Body item:", body_item)return {"message": "success"}

router模块化

items.py from fastapi import APIRouter,Header, HTTPException,Depends
async def tokenUser(token: str = Header()):if token != "123456":return HTTPException(status_code=401, detail="token error")router = APIRouter(prefix="/items", tags=["信息"], dependencies=[Depends(tokenUser)])@router.get("/info")
async def info():return {"message": "Hello World"}   

user.pyfrom fastapi import APIRouterrouter = APIRouter(prefix="/user", tags=["用户"])@router.get("/list")
async def listUser():return {"message": "获取到用户信息"}

main.py 

from fastapi import FastAPI
# 引入模块 在同一个目录下 直接 import 引入
# 引入模块 在其他目录下 需要到处目录名称 列入: form router import user,items
import user, items
app = FastAPI()


app.include_router(user.router)
app.include_router(items.router)