开发日志

#使用该装饰器可以对定义的模型中的成员进行类型校验
#使用正则表达式可以先定用户可以输入的合法字符
@field_validator("password")
    def validate_password(cls, value: str) -> str:
        pattern = r"^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)[\S]{8,20}$"
        if not re.fullmatch(pattern, value):
            raise ValueError(
                "密码必须8~20字符，且包含至少一个大写字母、小写字母和数字"
            )
        return value

@app.websocket
@router.add_api_websocket_route
#token的另一种解法？或者双层保险。登录后的某些会话可以使用

路径路由
资源使用
权限与JWT
websocket
后面的许多可等到完成全部以后进行复盘

@app.post("upload_file/{path_var}",summary="上传文件")
async def upload_file(*,
                      file: UploadFile,#请求体，声明数据模型的类
                      path_var: str | None = None,#路径变量
					  code: str| None = Query(None,min_length=4,max_length=8,alais="token")#路径参数
):
    file_local = await save_files(file)
    return{
        "file_name":file.filename,
        "content_type":file.content_type,
        "path_var":path_var,
        "code":code,
        "file_local":file_local
    }

async def save_files(file):
    path = "files"
    res = await file.read()
    hash_name = hashlib.md5(file.filename.encode()).hexdigest()[:8]
    file_name = f"{fashed_name}.{file.filename.rsplit('.',1)[-1]}"
    full_file = f"{path}/{file_name}"
    with open(full_file,"wb") as f:
        f.write(res)
    return full_file

#RBAC，基于角色的权限控制
ALL_USERS={
    'jack':['admin','user'],
    'rose':['admin','user'],
    'tom':['user'],
    'jerrt':['user']
}

ROLE_PERMISSIONS = {
    'admin':['upload'],
    'users':['visit','download'],
}

#模拟登录，获取Cookie
@router.get("/login")
async def set_cookie(resp:Response,token: str):
    resp.set_cookie(key="user_name",value=token,expires=600)
    return {"message":"cookie set"}

#获取登录的用户名
def get_user_token(user_name: str| None=Cookie(default=None)):
    if user_name is None:
        raise HTTPException(status_code=status.HTTP_401,detail="需要用户名")
        return user_name

#获取他的角色所拥有的所有操作权限
def get_role_permissions(role_name:list[str]):
    permissions = []
    for role in role_name:
        for perm in ROLE_PERMISSIONS[role]:
            permissions.append(perm)
    return permissions

#获取登录的用户[token](如果该用户存在)的所有角色
def get_user_permissions(token: str = Depends(get_user_token)):
    if token in ALL_USERS:
        return get_role_permissions(ALL_USERS[token])
    return None

#检测用户是否拥有执行该操作的权限
def check_user(security_scopes:SecurityScopes,user_permission:str=Depends(get_user_permissions)):
    for scope in security_scopes.scopes:
        if scope not in user_permission:
            raise HTTPException(status_code=status.HTTP_401,detail="你没有权限！")
   
#将check_user嵌入具体的操作路由中
@router.post("/test",dependencies=[Secutiry(check_user,scopes=['upload'])])
''''''