生成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))
说明官方文档：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-达到重试上限
使用Jinja2Templates模板引擎推荐文章：【FastAPI】Static静态文件配置安装命令：pip3installjinja2路径：\routes\index.pyfromfastapiimportFastAPI,Requestfromfastapi.responsesimportHTMLResponsefromfastapi.staticfilesimportStaticFilesfromfastapi.templatingimportJinja2Templatesapp=FastAPI()app.mount("/static",StaticFiles(directory="static"),name="static")#jscss等静态资源存放的目录templates=Jinja2Templates(directory="templates")#模板html存放的目录@app.get("/index",response_class=HTMLResponse)asyncdefread_item(request:Request):returntemplates.TemplateResponse(name='index.html',context={'request':request,#必要参数'data':{'title':'模板'},#渲染给前端的数据})main.js文件路径\static\main.jsconsole.log('加载成功')index.html模板路径：\templates\index.html<!DOCTYPEhtml><htmllang="en"><head><metacharset="UTF-8"><scriptsrc="{{url_for('static',path='main.js')}}"></script><title>Title</title></head><body><h1>{{data.title}}</h1></body></html>程序主文件fromfastapiimportFastAPIfromfastapi.staticfilesimportStaticFilesfromroutes.indeximportapplicationimportuvicornapp=FastAPI()#注意是在FastAPI()实例上挂载静态文件路径app.mount(path='/static',#网页的路径app=StaticFiles(directory='./static'),#静态文件目录的路径name='static')app.include_router(application,prefix='',tags=['首页'])if__name__=='__main__':uvicorn.run(app)
—-思路来自志远大佬什么是TLS说道ja3指纹，必然是要知道TLS的安全传输层协议（TLS）用于在两个通信应用程序之间提供保密性和数据完整性。—-来自百度百科什么是ja3官方链接：https://engineering.salesforce.com/tls-fingerprinting-with-ja3-and-ja3s-247362855967大概意思就是：JA3和JA3S作为一种方法来识别客户端和服务器之间的TLS协议，可以帮助对特定客户端与其服务器之间的某些指纹信息，来识别恶意通信。ja3测试使用网站：https://ja3er.com/jsonrequests、httpx、aiohttp的结果，这里就不做其他IP、UA等测试了，结果都是以知的一样查看TLS/SSL指纹信息过滤条件tls.handshake.version==0x00303使用的软件：鲨鱼魔改OpenSSL并编译把编译好的libcrypto-1_1-x64.dll、libssl-1_1-x64.dll使用curl和python测试curl测试Python测试
说明测试用例在项目中是不可缺失的一部分FastAPI的测试用例需要安装pipinstallpytest包含测试用例的文件名格式一般为：test_xx.py使用编写两个接口importuvicornfromrandomimportrandintfromtypingimportOptionalfromfastapiimportFastAPI,Headerapp=FastAPI()@app.get('/random')asyncdefget_random():"""获取一个随机数"""returnrandint(1,10)@app.get('/ua')asyncdefget_ua(user_agent:Optional[str]=Header(None,convert_underscores=True)):returnuser_agentif__name__=='__main__':uvicorn.run(app)编写测试用例文件需要导入FastAPI()实例化的对象fromfastapi.testclientimportTestClientfromrunimportappclient=TestClient(app=app)deftest_app_random():res=client.get('/random')assertres.status_code==200asserttype(res)==intdeftest_app_ua():res=client.get('/ua')assertres.status_code==200直接再命令行使用pytest启动测试
说明可以先返回响应，然后服务器再执行任务FastAPI的BackgroundTasks只能实现简单的后台任务，如需要开启进程等，就需要更强大的工具Celery使用后台任务importuvicornfromfastapiimportFastAPI,BackgroundTasksapp=FastAPI()#异步&同步函数红豆可以asyncdefwrite_notification(email:str,message):withopen('./log.txt',mode='w')asfile:file.write(f'notificationfor{email}:{message}')@app.post("/send-notification/{email}")asyncdefsend_notification(email:str,background_tasks:BackgroundTasks):#会自动识别为后台任务background_tasks.add_task(write_notification,email,message="somenotification")return{"message":"在后台发送通知"}if__name__=='__main__':uvicorn.run(app)依赖使用后台任务importuvicornfromfastapiimportFastAPI,BackgroundTasks,Dependsapp=FastAPI()asyncdefwrite_log(email:str):message="somenotification"withopen('./log.txt',mode='w')asfile:file.write(f'notificationfor{email}:{message}')#异步&同步函数红豆可以asyncdefadd_task(email:str,background_tasks:BackgroundTasks):background_tasks.add_task(write_log,email)@app.post("/send-notification/{email}")asyncdefsend_notification(q:str=Depends(add_task)):#使用依赖添加后台任务return{"message":"在后台发送通知"}if__name__=='__main__':uvicorn.run(app)
说明本文仅供学习交流，严禁用于非法用途，文章如有不当可联系博主删除网址：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']可以简单一点直接复制函数就行测试结果一样
什么是跨域资源共享是浏览器资源安全的一个方案，使开发的接口，不会被其他网站滥用，需要满足你指定的条件的网站才能使用条件可以指定：域，请求方法，头部信息域的格式：协议://域名:端口（https://bigdataboy.cn、https://bigdataboy.cn:433）请求方法：GET、POST、PUT…头部信息：指请求的头部信息使用importuvicornfromfastapiimportFastAPIfromfastapi.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"}if__name__=='__main__':uvicorn.run(app)
什么是中间件中间件是一个函数请求达到响应逻辑之前会经过一层或多层中间件，响应结果返回客户端也会经过一层或多层中间件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()
什么是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获取当前用户信息
什么是OAuth2认证规范：共同准守的规范，需要学习这种规范、流程OAuth2是一个规范，它定义了几种处理身份认证和授权的方法。基本的规范结构不标准的代码，只是讲解规范需要校验token的接口，把OAuth2PasswordBearer实例化对象作为依赖就好。fromfastapi.securityimportOAuth2PasswordBearer,OAuth2PasswordRequestForm"""1、OAuth2PasswordBearer只是接收URL作为参数的一个类，\客户端会向该URL发送用于请求token的参数（username、password...）2、OAuth2PasswordBearer不会创建相应的URL路径操作，\只是指明客户端用来请求Token的URL地址3、把OAuth2PasswordBearer的实例作为依赖，\FastAPI会检查请求的Authorization头信息，如果没有找到会返回401状态码（UNAUTHORIZED）"""#实例化OAuth2PasswordBearer指明请求Token地址oauth2_scheme=OAuth2PasswordBearer(tokenUrl='token')#http://127.0.0.1:8000/token#请求token时具体的处理@app.post('/token')#对应上面的OAuth2PasswordBearer实例化参数asyncdefget_token(form_data:OAuth2PasswordRequestForm=Depends()):"""这里使用了OAuth2规范的表单类""""""OAuth2规范的Token返回结构"""return{'access_token':user.username,'token_type':'bearer'}#使用oauth2_scheme作为依赖，会自动校验请求的Authorization头信息asyncdefget_user_info(user_token:str=Depends(oauth2_scheme)):returnuser_token#使用get_user_info作为依赖，直接请求会返回401状态码（UNAUTHORIZED）@app.post('/user/me')asyncdefread_user_me(user:str=Depends(get_user_info)):returnuser实例项目中不会这样使用（便于理解），项目更多的是使用JsonWebToken认证模式importuvicornfrompydanticimportBaseModelfromfastapiimportDepends,FastAPI,HTTPException,statusfromfastapi.securityimportOAuth2PasswordBearer,OAuth2PasswordRequestFormapp=FastAPI()fake_user_db={#模拟用户数据库'Bob':{'username':'Bob','password':'123456'},'Mary':{'username':'Mary','password':'987654'}}classUser(BaseModel):username:strpassword:str#实例化OAuth2PasswordBearer类，指明请求token接口地址oauth2_scheme=OAuth2PasswordBearer(tokenUrl='token')@app.post('/token')#获取token的接口具体方法asyncdefget_token(form_data:OAuth2PasswordRequestForm=Depends()):user_dict=fake_user_db.get(form_data.username)#数据库获取用户ifnotuser_dict:#检查用户是否存在raiseHTTPException(status_code=status.HTTP_400_BAD_REQUEST,detail='Incorrectusernameorpassword')user=User(**user_dict)ifform_data.password!=user.password:#检查密码是否正确raiseHTTPException(status_code=status.HTTP_400_BAD_REQUEST,detail='Incorrectusernameorpassword')"""access_token是返回的token（这里是为了方便，所以使用username作用token）token_type是对应实例化OAuth2Password'Bearer'"""return{'access_token':user.username,'token_type':'bearer'}#依赖oauth2_scheme自动校验Authorization头信息asyncdefget_user_info(user_token:str=Depends(oauth2_scheme)):returnUser(**fake_user_db[user_token])#获取token的用户详细信息@app.post('/user/me')asyncdefread_user_me(user:User=Depends(get_user_info)):returnuserif__name__=='__main__':uvicorn.run(app)接口测试页面直接请求，验证失败获取Token获取Token用户信息直接请求获取Token接口（@app.post(‘/token’)）