编程杂谈

说明TypeScript支持与JavaScript几乎相同的数据类型,此外还提供了实用的枚举类型布尔值letmsg:boolean=true;数字TypeScript里的所有数字都是浮点数。这些浮点数的类型是number。letn1:number=12;//十进制letn2:number=0b1010;//二进制letn3:number=0o12;//八进制letn4:number=0xa;//十六进制字符串单引号双引号都可以lets:string='xxx';undefined和null它们的本身的类型用处不是很大letund:undefined=undefined;letnu:null=null;数组TypeScript像JavaScript一样可以操作数组元素。有两种方式可以定义数组。第一种,let变量名:类型[]=[];letarr:number[]=[1,2,3,4];第二种泛型定义,Array<类型>letarr:Array<number>=[1,2,3,4];元组元组类型允许表示一个已知元素数量和类型的数组lettu:[number,string,boolean]=[1,'咪咪',true];枚举enum类型是对JavaScript标准数据类型的一个补充enumColor{Red,//0Green=10,Blue//11}letcolor:Color=Color.Blue;console.log(Color)//0any当不知定义什么类型比较合适时使用leta:any;a=12;a='12';letb:any=12;b='a';console.log(a,b)//'12''a'object所有对象的基类leto:object={'a':12};letoo:object=newString('oo');联合类型表示取值为多种类型中的一种functionfun(s:number|string):number|string{returns}类型断言当使用联合类型时,编译器可能不知道,现在我们处理的时候什么类型,而出现报错,因此就需要告诉编译现在处理的是什么。/*两种定义方式:第一种:(<类型>变量)第二种:(变量as类型)*/functionfun(s:number|string):number{//返回参数的长度if((<string>s).length){return(<string>s).length;}else{return(sasnumber).toString().length;}}类型推断当不定义类型是,TypeScript会自动推断类型,如果只是定义,不初始化变量,则类型为anyletn=123;//推断类型为number//n='a';//报错letm;//推断类型为anym=12;m='a';

编程杂谈

生成package.json文件终端运行npminit-ypackage.json文件{"name":"code","version":"1.0.0","description":"","main":"index.js","scripts":{"test":"echo\"Error:notestspecified\"&&exit1"},"keywords":[],"author":"","license":"ISC",}生成tsconfig.json文件终端运行,生成ts配置文件tsc--init创建html文件新建目录public创建html文件index.html<!DOCTYPEhtml><htmllang="en"><head><metacharset="UTF-8"><metahttp-equiv="X-UA-Compatible"content="IE=edge"><metaname="viewport"content="width=device-width,initial-scale=1.0"><title>Document</title></head><body></body></html>创建ts文件新建目录src创建ts文件main.ts(()=>{alert('打包成功啦~~~~')})()创建webpack.config.js文件新建目录./build创建Js文件webpack.config.jsconst{CleanWebpackPlugin}=require('clean-webpack-plugin')constHtmlWebpackPlugin=require('html-webpack-plugin')constpath=require('path')constisProd=process.env.NODE_ENV==='production'//是否生产环境functionresolve(dir){returnpath.resolve(__dirname,'..',dir)}module.exports={mode:isProd?'production':'development',//生产环境还是开发环境entry:{app:'./src/main.ts'//程序主目录},output:{//打包输出配置项path:resolve('dist'),//打包结果在dist目录下filename:'[name].[contenthash:8].js'//打包后的js文件名格式},module:{rules:[{//对src目录下的ts文件编译操作test:/\.tsx?$/,use:'ts-loader',include:[resolve('src')]}]},plugins:[newCleanWebpackPlugin({//把以前打包的js清除}),newHtmlWebpackPlugin({template:'./public/index.html'//当前目录的html打包})],resolve:{extensions:['.ts','.tsx','.js']//引入这类文件可以不写扩展名},devtool:isProd?'cheap-module-source-map':'eval-cheap-module-source-map',//提示错误信息devServer:{host:'localhost',//主机名port:8081,open:true//自动打开浏览器},}下载依赖npminstall-Dtypescriptnpminstall-Dwebpackwebpack-cliwebpack-dev-server//打包html和清除之前打包的jsnpminstall-Dhtml-webpack-pluginclean-webpack-plugin//对ts文件进行编译处理npminstall-Dts-loader#涉及跨平台的命令npminstall-Dcross-env配置打包命名//开发环境打包"dev":"cross-envNODE_ENV=developmentwebpack-dev-server--configbuild/webpack.config.js",//生产环境打包"build":"cross-envNODE_ENV=productionwebpack--configbuild/webpack.config.js"运行打包//开发环境打包测试npmrundev//生产环境打包npmrunbuild开发环境打包生产环境打包

编程杂谈

什么是TypeScriptTypeScript是一种由微软开发的开源、跨平台的编程语言。它是JavaScript的超集,最终会被编译为JavaScript代码。2012年10月,微软发布了首个公开版本的TypeScript,2013年6月19日,在经历了一个预览版之后微软正式发布了正式版TypeScriptTypeScript的作者是安德斯·海尔斯伯格,C#的首席架构师。它是开源和跨平台的编程语言。TypeScript扩展了JavaScript的语法,所以任何现有的JavaScript程序可以运行在TypeScript环境中。TypeScript是为大型应用的开发而设计,并且可以编译为JavaScript。TypeScript是JavaScript的一个超集,主要提供了类型系统和对ES6+的支持,它由Microsoft开发,代码开源于GitHub上TypeScript三大特点始于JavaScript,归于JavaScriptTypeScript可以编译出纯净、简洁的JavaScript代码,并且可以运行在任何浏览器上、Node.js环境中和任何支持ECMAScript3(或更高版本)的JavaScript引擎中。强大的类型系统类型系统允许JavaScript开发者在开发JavaScript应用程序时使用高效的开发工具和常用操作比如静态检查和代码重构。先进的JavaScriptTypeScript提供最新的和不断发展的JavaScript特性,包括那些来自2015年的ECMAScript和未来的提案中的特性,比如异步功能和Decorators,以帮助建立健壮的组件TypeScript快速上手安装TypeScript首选安装好node环境下载地址:https://nodejs.org/zh-cn/命令行运行如下命令,全局安装TypeScriptnpminstall-gtypescript安装完成后,在控制台运行如下命令,检查安装是否成功tsc-VTypeScript程序src/main.ts(()=>{functionhello(name:string){return"你好啊!"+name;}console.log(hello('NoNo'))})()手动编译手动编译,因为是.ts文件,文件不能被浏览器运行,需要编译成js文件在终端中打开,输入tscmain.ts,自动生成编译好的mian.js文件vscode自动编译命令行生成配置文件tsconfig.jsontsc--init修改tsconfig.json配置"outDir":"./src",//指定编译输出目录"strict":false,//关闭严格模式依次点击启动监视任务,当文件保存了改变,就会自动编译终端->运行任务->所有任务->监视tsconfig.json类型注解TypeScript的带来的高级功能//基本类型的注解functionhello(name:string){return"你好啊!"+name;}接口实现复杂注解使用接口,实现复杂注解interfacePerson{firstName:string,lastName:string}functionhello(name:Person){return"你好啊!"+name.firstName+'·'+name.lastName;}letuser={firstName:'No',lastName:'No'}console.log(hello(user))类TypeScript支持JavaScript的新特性,比如支持基于类的面向对象编程。classPerson{firstName:string//属性lastName:string//构造函数constructor(firstName:string,lastName:string){this.firstName=firstNamethis.lastName=lastName}}functionhello(name:Person){return"你好啊!"+name.firstName+'·'+name.lastName;}letuser=newPerson('No','No')console.log(hello(user))

Python极客

说明官方文档:https://tenacity.readthedocs.io/en/latest/Tenacity是一个Apache2.0许可的通用重试库,用Python编写,用于简化将重试行为添加到几乎任何内容的任务。它起源于retrying的一个分支,遗憾的是retrying不再维护。Tenacity与retrying不兼容,但添加了重要的新功能并修复了许多长期存在的错误。特征通用装饰器API指定停止条件(即限制尝试次数)指定等待条件(即尝试之间的指数退避休眠)自定义对预期返回结果的重试停止、等待、重试条件任意组合协程重试使用上下文管理器重试代码块安装:pipinstalltenacity例子不加任何条件,一直重试到正确结果,才停止重试importrandomfromtenacityimportretry@retrydeffun():r=random.randint(1,3)print(f'当前{r}')assertr==2if__name__=='__main__':fun()-----------------------------------当前1当前3当前3当前2Processfinishedwithexitcode0停止条件stop_after_attempt比较常用,停止条件还有很多fromtenacityimportretryfromtenacity.stopimportstop_after_attempt#重试2次,停止@retry(stop=stop_after_attempt(2))deffun():pass等待条件fromtenacityimportretry#每次重试等待2秒fromtenacity.waitimportwait_fixed@retry(wait=wait_fixed(2))deffun():#每次重试随机等待1~2秒fromtenacity.waitimportwait_random@retry(wait=wait_random(min=1,max=2))deffun():#在重试分布式服务和其他远程端点时,结合固定等待和抖动(以帮助避免雷鸣般的群体)#固定基础等待3秒,随机增加0~2秒fromtenacity.waitimportwait_fixed,wait_random@retry(wait=wait_fixed(3)+wait_random(0,2))deffun():#生成一个等待链(等待时间的列表)等待时间相对于:[3,3,3,7,7,9]fromtenacity.waitimportwait_chain,wait_fixed@retry(wait=wait_chain(*[wait_fixed(3)foriinrange(3)]+[wait_fixed(7)foriinrange(2)]+[wait_fixed(9)]))deffun():wait_exponential重试分布式服务和其他远程端点时,等待时间呈指数增长multiplier:增长的指数【默认值1】min:最小等待时间【默认值0】max:最大等待时间【默认值sys.maxsize/2】importrandomfromloguruimportloggerfromtenacityimportretryfromtenacity.waitimportwait_exponential@retry(wait=wait_exponential(multiplier=1,min=4,max=10))deffun():r=random.randint(1,5)logger.info(f'当前{r}')assertr==6if__name__=='__main__':fun()-----------------------------------------------2022-02-2810:52:37.918|INFO|__main__:fun:10-当前22022-02-2810:52:41.930|INFO|__main__:fun:10-当前22022-02-2810:52:45.938|INFO|__main__:fun:10-当前12022-02-2810:52:49.949|INFO|__main__:fun:10-当前42022-02-2810:52:57.952|INFO|__main__:fun:10-当前32022-02-2810:53:07.965|INFO|__main__:fun:10-当前12022-02-2810:53:17.975|INFO|__main__:fun:10-当前42022-02-2810:53:27.977|INFO|__main__:fun:10-当前3stop_after_delay函数运行超过限制秒数,引发重试异常importrandomimporttimefromtenacityimportretryfromtenacity.waitimportstop_after_delay@retry(stop=stop_after_delay(2))deffun():r=random.randint(1,5)print(f'当前{r}')time.sleep(2)assertr==2print(f'对了')-----------------------------------tenacity.RetryError:RetryError[<Futureat0x260529a62b0state=finishedraisedAssertionError>]是否重试#AssertionError异常,触发重试fromtenacity.retryimportretry_if_exception_type@retry(retry=retry_if_exception_type(AssertionError))deffun():使用函数判断,是否重试importrandomfromloguruimportloggerfromtenacityimportretryfromtenacity.retryimportretry_if_result@retry(retry=retry_if_result(lambdax:x!=2))deffun():r=random.randint(1,5)logger.info(f'当前{r}')returnrif__name__=='__main__':fun()--------------------------------------------2022-02-2811:26:33.588|INFO|__main__:fun:10-当前52022-02-2811:26:33.588|INFO|__main__:fun:10-当前12022-02-2811:26:33.588|INFO|__main__:fun:10-当前32022-02-2811:26:33.588|INFO|__main__:fun:10-当前2组合重试条件importrandomfromloguruimportloggerfromtenacityimportretryfromtenacity.retryimportretry_if_result,retry_if_exception_type"""函数结果不可能为2和3-2重试-3触发AssertionError异常,重试"""@retry(retry=(retry_if_result(lambdax:x!=2)andretry_if_exception_type(AssertionError)))deffun():r=random.randint(1,5)logger.info(f'当前{r}')ifr==3:raiseAssertionError(f'ris3')returnrif__name__=='__main__':logger.info(f'fun{fun()}')协程重试importrandomimportasynciofromloguruimportloggerfromtenacityimportretryfromtenacity.waitimportwait_fixed@retry(wait=wait_fixed(2))asyncdeffun():r=random.randint(1,3)logger.info(f'当前{r}')assertr==2if__name__=='__main__':asyncio.run(fun())-----------------------------------------2022-02-2812:36:07.801|INFO|__main__:fun:11-当前32022-02-2812:36:09.805|INFO|__main__:fun:11-当前12022-02-2812:36:11.807|INFO|__main__:fun:11-当前32022-02-2812:36:13.814|INFO|__main__:fun:11-当前2统计数据函数.retry.statistics重试次数开始运行时间整个重试的耗时importrandomfromloguruimportloggerfromtenacityimportretryfromtenacity.retryimportretry_if_result@retry(reraise=True,retry=retry_if_result(lambdax:x!=2))deffun():r=random.randint(1,5)logger.info(f'当前{r}')returnrif__name__=='__main__':logger.info(f'fun{fun()}')logger.info(f'fun{fun.retry.statistics}')----------------------------------------------------2022-02-2811:43:33.558|INFO|__main__:fun:10-当前12022-02-2811:43:33.558|INFO|__main__:fun:10-当前12022-02-2811:43:33.558|INFO|__main__:fun:10-当前12022-02-2811:43:33.559|INFO|__main__:fun:10-当前32022-02-2811:43:33.559|INFO|__main__:fun:10-当前52022-02-2811:43:33.559|INFO|__main__:fun:10-当前42022-02-2811:43:33.559|INFO|__main__:fun:10-当前22022-02-2811:43:33.559|INFO|__main__:<module>:15-fun22022-02-2811:43:33.559|INFO|__main__:<module>:16-fun{'start_time':6972.546,'attempt_number':7,'idle_for':0,'delay_since_first_attempt':0.0}重试代码块重试代码块的逻辑包含的一个for循环中,然后包含一个上下文管理器,使需要重试的逻辑脱离函数importrandomfromloguruimportloggerfromtenacityimportRetrying,RetryError,stop_after_attempttry:#重试3次forattemptinRetrying(stop=stop_after_attempt(3)):#重试、等待、停止条件设置withattempt:#重试逻辑包含在上下文管理器r=random.randint(1,3)logger.info(f'r->{r}')assertr==2exceptRetryError:logger.error(f'超出重试次数')#超出重试次数处理逻辑pass----------------------------------------------------------2022-02-2813:08:13.491|INFO|__main__:<module>:10-r->32022-02-2813:08:13.492|INFO|__main__:<module>:10-r->32022-02-2813:08:13.492|INFO|__main__:<module>:10-r->32022-02-2813:08:13.492|ERROR|__main__:<module>:13-超出重试次数异步重试代码块异步使用importrandomimportasynciofromloguruimportloggerfromtenacityimportAsyncRetrying,RetryError,stop_after_attemptasyncdeffun():try:asyncforattemptinAsyncRetrying(stop=stop_after_attempt(3)):#重试、等待、停止条件设置#这个with前不能带asyncwithattempt:#重试逻辑包含在上下文管理器r=random.randint(1,3)logger.debug(f'r->{r}')assertr==2exceptRetryError:logger.error('达到重试上限')passif__name__=='__main__':asyncio.run(fun())------------------------------------------------------------------2022-02-2813:09:17.247|DEBUG|__main__:fun:13-r->32022-02-2813:09:17.247|DEBUG|__main__:fun:13-r->12022-02-2813:09:17.247|DEBUG|__main__:fun:13-r->32022-02-2813:09:17.247|ERROR|__main__:fun:16-达到重试上限

编程杂谈

说明本文仅供学习交流,严禁用于非法用途,文章如有不当可联系博主删除网址:aHR0cHM6Ly9tLmN0eXVuLmNuL3dhcC9tYWluL2F1dGgvbG9naW4/cmVkaXJlY3Q9JTJGbXk=加密参数:password,comParam_seqCode,comParam_signature开始断点追栈找password加密处这里有个小技巧,先尝试看函数名,成的话就可以不用一层一层追了第一个login打断点,就发现了加密处//分析发现//加密函数是Object(l["c"])(a.value,Object(l["f"])(Object(l["g"])(s.value)))//其中Object(l["c"])为加密函数,它有两个参数第一个参数是密码a.value第二个参数是对邮箱进行了处理Object(l["f"])(Object(l["g"])(s.value))主要对邮箱的处理函数是Object(l["f"]),作用是保证第二个参数是'24位'不够'补0'多了'截取'分析加密逻辑复制整个js文件,发现是webpak打包的js,那就简单了//整个加密的核心是这行p.a.TripleDES.encrypt(e,d,l);//找到p变量是怎么来的,d=t("80e3"),l=t.n(d),s=t("3452")//s来自t("3452"),p=t.n(s)//p变量来自s,f=t("c466"),h=null,m=function(e){returnfunction(n){returnObject.prototype.toString.call(n)==="[object".concat(e,"]")}扒代码发现这是webpack打包过的,我们构造一个简单的加载器varfff;!function(e){varn={}functionf(t){if(n[t])returnn[t].exports;console.log('-->',t);varr=n[t]={i:t,l:!1,exports:{}};returne[t].call(r.exports,r,r.exports,f),r.l=!0,r.exports}fff=f;}({//函数})填入3452对象运行代码执行fff('3452'),发现报错,这是缺少代码在网站Js代码的加载器打断点,找到缺少的函数,然后复制粘贴代码补充完整,得到加密对象改写加密逻辑然后测试运行继续找comParam_seqCode、comParam_signature看着主要的逻辑//h.getTimestampOffset()的逻辑f是h.getTimestampOffset=function(){returnlocalStorage.getItem("timestampOffset")||f}//其他关键的点`u['k']`、`o()`向上翻,找到u['k']、o()定义点扒取代码,跟上面一样,这里的u['k']可以简单一点直接复制函数就行测试结果一样

Python极客

什么是中间件中间件是一个函数请求达到响应逻辑之前会经过一层或多层中间件,响应结果返回客户端也会经过一层或多层中间件FastAPI中间件开发自定义中间件importtimeimportuvicornfromfastapiimportFastAPIfromfastapi.requestsimportRequestfromfastapi.responsesimportResponseapp=FastAPI()@app.middleware('http')#中间件类型现在只支持http类型asyncdefadd_process_time_header(request:Request,call_next):"""计算请求响应时间:paramrequest:请求:paramcall_next:请求处理回调:return:"""start_time=time.time()res:Response=awaitcall_next(request)process_time=time.time()-start_timeres.headers['X-Process-Time']=str(process_time)returnres@app.get('/middleware')defmiddleware():return'没有处理逻辑,响应成功'if__name__=='__main__':uvicorn.run(app)提供的中间件CORSMiddleware跨域资源共享中间件fromfastapiimportFastAPIfromfastapi.middleware.corsimportCORSMiddlewareapp=FastAPI()app.add_middleware(CORSMiddleware,allow_origins=[#允许访问的域域-->协议:域名:端口'http://127.0.0.1','http://127.0.0.1:8000',],allow_methods=['*','GET'],#*可以通配allow_headers=['*'],#头部allow_credentials=False,#HTTPS证书allow_origin_regex=None,#正则表达式匹配'https://.*\.example\.orgexpose_headers=[],#指明可以被浏览器访问的响应头max_age=600#设定浏览器缓存CORS响应的最长时间,单位是秒。默认为600)@app.get("/")asyncdefmain():return{"message":"HelloWorld"}HTTPSRedirectMiddleware强制所有传入请求必须是https或wss如果不是https则会自动跳转到https,如果网站没有配置https,则报错h11._util.RemoteProtocolError:illegalrequestlineWARNING:InvalidHTTPrequestreceived.fromfastapiimportFastAPIfromfastapi.middleware.httpsredirectimportHTTPSRedirectMiddlewareapp=FastAPI()app.add_middleware(HTTPSRedirectMiddleware)@app.get("/")asyncdefmain():return{"message":"HelloWorld"}TrustedHostMiddleware强制所有传入请求都具有正确设置的Host标头,以防止HTTP主机标头攻击。如果不包含,则响应400状态码,并返回InvalidhostheaderfromfastapiimportFastAPIfromfastapi.middleware.trustedhostimportTrustedHostMiddlewareapp=FastAPI()app.add_middleware(TrustedHostMiddleware,allowed_hosts=["bigdataboy.cn","*.bigdataboy.cn"]#允许的hosts列表)@app.get("/")asyncdefmain():return{"message":"HelloWorld"}GZipMiddleware处理包含”gzip”在Accept-Encoding标头中的任何请求的GZip响应。fromfastapiimportFastAPIfromfastapi.middleware.gzipimportGZipMiddlewareapp=FastAPI()app.add_middleware(GZipMiddleware,minimum_size=1000)#不要GZip响应小于此最小字节大小。默认为500.@app.get("/")asyncdefmain():return{"message":"HelloWorld"}带yield关键字依赖,依赖中的退出代码将在执行中间件后执行defget_db():db=dbSession()try:yielddbfinally:db.close()

Python极客

什么是JsonWebTokens(JWT)请求流程首先浏览器向服务器发送请求得到token然后浏览器每次请求都自动在头部带上tokentoken的组成token是一些信息(用户名,过期时间…)的加密结果,加密的秘钥保存在服务器,因此只要秘钥不被泄漏,就认为是安全的。FastAPI的JWT哈希密码把用户的明文密码,加密保存在数据库,即使密码泄漏,也不知道用户真正的密码。推荐的算法是「Bcrypt」:pip3installpasslib[bcrypt]frompasslib.contextimportCryptContext#创建哈希上下文pwd_context=CryptContext(schemes=["bcrypt"],deprecated="auto")#校验密文明文defverify_password(plain_password,hashed_password):returnpwd_context.verify(plain_password,hashed_password)#加密明文defget_password_hash(password):returnpwd_context.hash(password)if__name__=='__main__':password_hash=get_password_hash('123456')#加密结果是动态的每次不一样,但是验证是一样的print(password_hash)print(verify_password('123456',password_hash))token生成&校验逻辑说明需要加密使用的秘钥加密算法token过期时间#opensslrand-hex32SECRET_KEY="12b2e365a1b19051f115e46e8dfd7200e63510319a791bcb2dcf605626e1aa0c"ALGORITHM="HS256"ACCESS_TOKEN_EXPIRE_MINUTES=30token生成与校验安装模块:pipinstallpython-jose[cryptography]fromtypingimportOptionalfromjoseimportJWTError,jwtfromdatetimeimportdatetime,timedelta#opensslrand-hex32SECRET_KEY="12b2e365a1b19051f115e46e8dfd7200e63510319a791bcb2dcf605626e1aa0c"ALGORITHM="HS256"ACCESS_TOKEN_EXPIRE_MINUTES=30defcreate_access_token(user:dict,expires_delta:Optional[timedelta]=None):to_encode=user.copy()#浅拷贝:深拷贝父对象(一级目录),子对象(二级目录)不拷贝,还是引用ifexpires_delta:#能通过参数指定过期时间expire=datetime.utcnow()+expires_deltaelse:expire=datetime.utcnow()+timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)to_encode.update({"exp":expire})#加入过期时间encoded_jwt=jwt.encode(to_encode,SECRET_KEY,algorithm=ALGORITHM)#加密信息得到tokenreturnencoded_jwtif__name__=='__main__':access_token_expires=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)#生成时间格式0:30:00token=create_access_token(#生成Token{'username':'Bob'},expires_delta=access_token_expires)print(f'user-->token:{token}')to_decode=jwt.decode(token,SECRET_KEY,algorithms=[ALGORITHM])#解密tokenprint(f'token-->user{to_decode}')案例此案例结合上篇的基于PasswordBearerToken的OAuth2认证获取TokenimportuvicornfromtypingimportOptionalfrompydanticimportBaseModelfromdatetimeimportdatetime,timedeltafromfastapiimportDepends,FastAPI,HTTPException,statusfromfastapi.securityimportOAuth2PasswordBearer,OAuth2PasswordRequestFormfromjoseimportJWTError,jwt#token使用frompasslib.contextimportCryptContext#哈希密码#opensslrand-hex32SECRET_KEY="12b2e365a1b19051f115e46e8dfd7200e63510319a791bcb2dcf605626e1aa0c"ALGORITHM="HS256"ACCESS_TOKEN_EXPIRE_MINUTES=30fake_user_db={#模拟用户数据库'Bob':{'username':'Bob','hash_password':'$2b$12$Qm6i.pPxCM/Kc672T0GJZOr6Wnq2YkjZm.UMk1O9abq.8fx3fas52'},#明文123456'Mary':{'username':'Mary','hash_password':'$2b$12$Qm6i.pPxCM/Kc672T0GJZOr6Wnq2YkjZm.UMk1O9abq.8fx3fas52'},#明文123456}classUser(BaseModel):username:strhash_password:strclassToken(BaseModel):access_token:strtoken_type:str='bearer'pwd_context=CryptContext(schemes=["bcrypt"],deprecated="auto")#实例化OAuth2PasswordBearer类,指明请求token接口地址oauth2_scheme=OAuth2PasswordBearer(tokenUrl='/jwt/token')app=FastAPI()#验证密码defverify_password(plain_password:str,hash_password)->bool:returnpwd_context.verify(plain_password,hash_password)defget_password_hash(password:str):returnpwd_context.hash(password)defget_user(db,username:str)->User:ifusernameindb:user_dict=db[username]returnUser(**user_dict)defauthenticate_user(fake_db,username:str,password:str):user=get_user(fake_db,username)ifnotuser:returnFalseifnotverify_password(password,user.hash_password):returnFalsereturnuserdefcreate_access_token(user:dict,expires_delta:Optional[timedelta]=None):to_encode=user.copy()#浅拷贝:深拷贝父对象(一级目录),子对象(二级目录)不拷贝,还是引用ifexpires_delta:#能通过参数指定过期时间expire=datetime.utcnow()+expires_deltaelse:expire=datetime.utcnow()+timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)to_encode.update({"exp":expire})#加入过期时间encoded_jwt=jwt.encode(to_encode,SECRET_KEY,algorithm=ALGORITHM)#加密信息得到tokenreturnencoded_jwt@app.post('/jwt/token')#获取token的接口具体方法asyncdefget_token(form_data:OAuth2PasswordRequestForm=Depends()):user=authenticate_user(#获取用户对象fake_db=fake_user_db,username=form_data.username,password=form_data.password)ifnotuser:raiseHTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail='Incorrectusernameorpassword',headers={'WWW-Authenticate':'Bearer'})#规范access_token_expires=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)access_token=create_access_token(#生成Tokenuser=user.dict(),expires_delta=access_token_expires)returnToken(access_token=access_token,token_type='bearer')if__name__=='__main__':uvicorn.run(app)获取当前用户信息asyncdefget_current_user(token:str=Depends(oauth2_scheme)):credentials_exception=HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail="Couldnotvalidatecredentials",headers={"WWW-Authenticate":"Bearer"},)try:#解密token获取用户payload=jwt.decode(token,SECRET_KEY,algorithms=[ALGORITHM])username:str=payload.get('username')ifusernameisNone:raisecredentials_exceptiontoken_data=usernameexceptJWTError:returncredentials_exceptionuser=get_user(fake_user_db,token_data)ifuserisNone:raisecredentials_exceptionreturnuser@app.post('/jwt/me')asyncdefread_users_me(current_user:User=Depends(get_current_user)):returncurrent_user获取Token获取当前用户信息