搭建（學(xué)習(xí)階段2:權(quán)限校驗(yàn)）)
在gRPC的權(quán)限校驗(yàn)中寫一個(gè)類然后繼承g(shù)rpc.ServerInterceptor即可使用權(quán)限校驗(yàn)每隔函數(shù)在調(diào)用前都會(huì)調(diào)用這個(gè)函數(shù)這個(gè)函數(shù)可用于權(quán)限校驗(yàn)或是一些數(shù)據(jù)準(zhǔn)備等操作。文章目錄權(quán)限校驗(yàn)示例程序服務(wù)端代碼客戶端代碼帶Login的權(quán)限管理服務(wù)端代碼客戶端代碼權(quán)限校驗(yàn)示例程序服務(wù)端代碼importjsonfromconcurrentimportfuturesimportgrpc# 業(yè)務(wù)函數(shù)收到客戶端請(qǐng)求后返回問候消息。defsay_hello(request,_context):return{message:fHello,{request[name]}!}# 攔截器是 gRPC 的固定擴(kuò)展寫法可以在請(qǐng)求到達(dá)業(yè)務(wù)函數(shù)前做日志、鑒權(quán)等處理。# 這里為了保持簡(jiǎn)單只打印請(qǐng)求方法不做真正的權(quán)限校驗(yàn)。classLogInterceptor(grpc.ServerInterceptor):defintercept_service(self,next_step,handler_call_details):print(f收到請(qǐng)求{handler_call_details.method})returnnext_step(handler_call_details)defserver(envdev,address[::]:50051,max_workers2):啟動(dòng) gRPC 服務(wù)。 env運(yùn)行環(huán)境名稱當(dāng)前只用于打印提示。 address監(jiān)聽地址格式通常是“主機(jī):端口”。 max_workers線程池中的最大工作線程數(shù)。 print(f當(dāng)前運(yùn)行環(huán)境{env})# 這是 gRPC 創(chuàng)建服務(wù)端的固定寫法。grpc_servergrpc.server(futures.ThreadPoolExecutor(max_workersmax_workers),interceptors[LogInterceptor()])# 下面是本示例的 JSON 請(qǐng)求處理器。# 標(biāo)準(zhǔn) gRPC 項(xiàng)目通常會(huì)用 .proto 文件自動(dòng)生成這部分代碼。handlergrpc.unary_unary_rpc_method_handler(say_hello,request_deserializerlambdadata:json.loads(data.decode(utf-8)),response_serializerlambdaresponse:json.dumps(response).encode(utf-8),)# 注冊(cè)服務(wù)和方法這是手動(dòng)注冊(cè) RPC 方法的寫法。grpc_server.add_generic_rpc_handlers((grpc.method_handlers_generic_handler(hello.Greeter,{SayHello:handler}),))# 添加監(jiān)聽端口、啟動(dòng)服務(wù)、持續(xù)等待是 gRPC 服務(wù)啟動(dòng)的固定流程。grpc_server.add_insecure_port(address)grpc_server.start()print(fgRPC 服務(wù)已啟動(dòng)監(jiān)聽地址{address})grpc_server.wait_for_termination()# 程序從這里開始執(zhí)行調(diào)用 server() 啟動(dòng)服務(wù)。if__name____main__:server()客戶端代碼importjsonimportgrpc# 連接地址必須和 server.py 中的端口一致。withgrpc.insecure_channel(127.0.0.1:50051)aschannel:# 指定要調(diào)用的服務(wù)方法/服務(wù)名/方法名。say_hellochannel.unary_unary(/hello.Greeter/SayHello,# 發(fā)送請(qǐng)求前Python 字典轉(zhuǎn)換為 JSON 字節(jié)。request_serializerlambdadata:json.dumps(data).encode(utf-8),# 收到響應(yīng)后JSON 字節(jié)轉(zhuǎn)換為 Python 字典。response_deserializerlambdadata:json.loads(data.decode(utf-8)),)# 發(fā)送名稱并獲取服務(wù)端返回的問候消息。responsesay_hello({name:World})print(response[message])帶Login的權(quán)限管理使用帶login的權(quán)限管理核心在于AuthInterceptor這個(gè)類中當(dāng)method /auth.Auth/Login時(shí)直接執(zhí)行校驗(yàn)當(dāng)執(zhí)行其他的method的時(shí)候直接替換調(diào)用check_token這個(gè)函數(shù)校驗(yàn)token如果可以就通過handler.unary_unary(request, context)繼續(xù)執(zhí)行如果不行就通過context.abort(grpc.StatusCode.UNAUTHENTICATED, 請(qǐng)先登錄)直接返回不進(jìn)行后續(xù)的執(zhí)行服務(wù)端代碼importjsonfromconcurrentimportfuturesimportgrpc# 服務(wù)端保存登錄成功后的 Token。TOKENNone# 登錄接口不需要 Token登錄成功后生成并保存 Token。deflogin(request,_context):globalTOKENifrequest[username]!adminorrequest[password]!123456:return{success:False,message:用戶名或密碼錯(cuò)誤}TOKENtoken-123return{success:True,token:TOKEN,message:登錄成功}# 只有通過 Token 校驗(yàn)后才會(huì)執(zhí)行這個(gè)業(yè)務(wù)函數(shù)。defsay_hello(request,_context):return{message:fHello,{request[name]}!}# 將普通 Python 函數(shù)注冊(cè)成 gRPC 方法并處理 JSON 轉(zhuǎn)換。defjson_handler(function):returngrpc.unary_unary_rpc_method_handler(function,request_deserializerlambdadata:json.loads(data.decode(utf-8)),response_serializerlambdadata:json.dumps(data).encode(utf-8),)# 在業(yè)務(wù)函數(shù)執(zhí)行前檢查 Token。classAuthInterceptor(grpc.ServerInterceptor):defintercept_service(self,continuation,handler_call_details):methodhandler_call_details.method handlercontinuation(handler_call_details)ifhandlerisNone:returnNone# 登錄接口不能檢查 Token否則客戶端無(wú)法登錄。ifmethod/auth.Auth/Login:returnhandler# 其他接口都必須攜帶正確的 Token。defcheck_token(request,context):ifrequest.get(token)!TOKEN:context.abort(grpc.StatusCode.UNAUTHENTICATED,請(qǐng)先登錄)returnhandler.unary_unary(request,context)# Token 正確繼續(xù)執(zhí)行原來(lái)的業(yè)務(wù)函數(shù)# 用權(quán)限檢查函數(shù)替換原來(lái)的業(yè)務(wù)處理函數(shù)。returnhandler._replace(unary_unarycheck_token)defserver(address[::]:50051,max_workers2):啟動(dòng) gRPC 服務(wù)。# 創(chuàng)建服務(wù)并安裝 AuthInterceptor 權(quán)限攔截器grpc_servergrpc.server(futures.ThreadPoolExecutor(max_workersmax_workers),interceptors[AuthInterceptor()])# 注冊(cè)登錄接口grpc_server.add_generic_rpc_handlers((grpc.method_handlers_generic_handler(auth.Auth,{Login:json_handler(login)},),# 注冊(cè)需要登錄的問候接口。grpc.method_handlers_generic_handler(hello.Greeter,{SayHello:json_handler(say_hello)},),))grpc_server.add_insecure_port(address)grpc_server.start()print(fgRPC 服務(wù)已啟動(dòng)監(jiān)聽地址{address})grpc_server.wait_for_termination()if__name____main__:server()客戶端代碼importjsonimportgrpc# 連接地址必須和 server.py 中的端口一致。withgrpc.insecure_channel(127.0.0.1:50051)aschannel:# 第一步調(diào)用登錄接口獲取服務(wù)端保存的 Token。loginchannel.unary_unary(/auth.Auth/Login,request_serializerlambdadata:json.dumps(data).encode(utf-8),response_deserializerlambdadata:json.loads(data.decode(utf-8)),)login_resultlogin({username:admin,password:123456})print(login_result[message])# 第二步調(diào)用需要登錄的 SayHello 接口。say_hellochannel.unary_unary(/hello.Greeter/SayHello,request_serializerlambdadata:json.dumps(data).encode(utf-8),response_deserializerlambdadata:json.loads(data.decode(utf-8)),)# 登錄成功后把 Token 和名稱一起發(fā)送給服務(wù)端。responsesay_hello({name:World,token:login_result[token]})print(response[message])# 第三步不發(fā)送 Token模擬未登錄調(diào)用。try:say_hello({name:World})exceptgrpc.RpcErroraserror:print(f未登錄{error.details()}) gRPC微服務(wù)搭建（學(xué)習(xí)階段2:權(quán)限校驗(yàn)）
在gRPC的權(quán)限校驗(yàn)中寫一個(gè)類然后繼承g(shù)rpc.ServerInterceptor即可使用權(quán)限校驗(yàn)每隔函數(shù)在調(diào)用前都會(huì)調(diào)用這個(gè)函數(shù)這個(gè)函數(shù)可用于權(quán)限校驗(yàn)或是一些數(shù)據(jù)準(zhǔn)備等操作。文章目錄權(quán)限校驗(yàn)示例程序服務(wù)端代碼客戶端代碼帶Login的權(quán)限管理服務(wù)端代碼客戶端代碼權(quán)限校驗(yàn)示例程序服務(wù)端代碼importjsonfromconcurrentimportfuturesimportgrpc# 業(yè)務(wù)函數(shù)收到客戶端請(qǐng)求后返回問候消息。defsay_hello(request,_context):return{message:fHello,{request[name]}!}# 攔截器是 gRPC 的固定擴(kuò)展寫法可以在請(qǐng)求到達(dá)業(yè)務(wù)函數(shù)前做日志、鑒權(quán)等處理。# 這里為了保持簡(jiǎn)單只打印請(qǐng)求方法不做真正的權(quán)限校驗(yàn)。classLogInterceptor(grpc.ServerInterceptor):defintercept_service(self,next_step,handler_call_details):print(f收到請(qǐng)求{handler_call_details.method})returnnext_step(handler_call_details)defserver(envdev,address[::]:50051,max_workers2):啟動(dòng) gRPC 服務(wù)。 env運(yùn)行環(huán)境名稱當(dāng)前只用于打印提示。 address監(jiān)聽地址格式通常是“主機(jī):端口”。 max_workers線程池中的最大工作線程數(shù)。 print(f當(dāng)前運(yùn)行環(huán)境{env})# 這是 gRPC 創(chuàng)建服務(wù)端的固定寫法。grpc_servergrpc.server(futures.ThreadPoolExecutor(max_workersmax_workers),interceptors[LogInterceptor()])# 下面是本示例的 JSON 請(qǐng)求處理器。# 標(biāo)準(zhǔn) gRPC 項(xiàng)目通常會(huì)用 .proto 文件自動(dòng)生成這部分代碼。handlergrpc.unary_unary_rpc_method_handler(say_hello,request_deserializerlambdadata:json.loads(data.decode(utf-8)),response_serializerlambdaresponse:json.dumps(response).encode(utf-8),)# 注冊(cè)服務(wù)和方法這是手動(dòng)注冊(cè) RPC 方法的寫法。grpc_server.add_generic_rpc_handlers((grpc.method_handlers_generic_handler(hello.Greeter,{SayHello:handler}),))# 添加監(jiān)聽端口、啟動(dòng)服務(wù)、持續(xù)等待是 gRPC 服務(wù)啟動(dòng)的固定流程。grpc_server.add_insecure_port(address)grpc_server.start()print(fgRPC 服務(wù)已啟動(dòng)監(jiān)聽地址{address})grpc_server.wait_for_termination()# 程序從這里開始執(zhí)行調(diào)用 server() 啟動(dòng)服務(wù)。if__name____main__:server()客戶端代碼importjsonimportgrpc# 連接地址必須和 server.py 中的端口一致。withgrpc.insecure_channel(127.0.0.1:50051)aschannel:# 指定要調(diào)用的服務(wù)方法/服務(wù)名/方法名。say_hellochannel.unary_unary(/hello.Greeter/SayHello,# 發(fā)送請(qǐng)求前Python 字典轉(zhuǎn)換為 JSON 字節(jié)。request_serializerlambdadata:json.dumps(data).encode(utf-8),# 收到響應(yīng)后JSON 字節(jié)轉(zhuǎn)換為 Python 字典。response_deserializerlambdadata:json.loads(data.decode(utf-8)),)# 發(fā)送名稱并獲取服務(wù)端返回的問候消息。responsesay_hello({name:World})print(response[message])帶Login的權(quán)限管理使用帶login的權(quán)限管理核心在于AuthInterceptor這個(gè)類中當(dāng)method /auth.Auth/Login時(shí)直接執(zhí)行校驗(yàn)當(dāng)執(zhí)行其他的method的時(shí)候直接替換調(diào)用check_token這個(gè)函數(shù)校驗(yàn)token如果可以就通過handler.unary_unary(request, context)繼續(xù)執(zhí)行如果不行就通過context.abort(grpc.StatusCode.UNAUTHENTICATED, 請(qǐng)先登錄)直接返回不進(jìn)行后續(xù)的執(zhí)行服務(wù)端代碼importjsonfromconcurrentimportfuturesimportgrpc# 服務(wù)端保存登錄成功后的 Token。TOKENNone# 登錄接口不需要 Token登錄成功后生成并保存 Token。deflogin(request,_context):globalTOKENifrequest[username]!adminorrequest[password]!123456:return{success:False,message:用戶名或密碼錯(cuò)誤}TOKENtoken-123return{success:True,token:TOKEN,message:登錄成功}# 只有通過 Token 校驗(yàn)后才會(huì)執(zhí)行這個(gè)業(yè)務(wù)函數(shù)。defsay_hello(request,_context):return{message:fHello,{request[name]}!}# 將普通 Python 函數(shù)注冊(cè)成 gRPC 方法并處理 JSON 轉(zhuǎn)換。defjson_handler(function):returngrpc.unary_unary_rpc_method_handler(function,request_deserializerlambdadata:json.loads(data.decode(utf-8)),response_serializerlambdadata:json.dumps(data).encode(utf-8),)# 在業(yè)務(wù)函數(shù)執(zhí)行前檢查 Token。classAuthInterceptor(grpc.ServerInterceptor):defintercept_service(self,continuation,handler_call_details):methodhandler_call_details.method handlercontinuation(handler_call_details)ifhandlerisNone:returnNone# 登錄接口不能檢查 Token否則客戶端無(wú)法登錄。ifmethod/auth.Auth/Login:returnhandler# 其他接口都必須攜帶正確的 Token。defcheck_token(request,context):ifrequest.get(token)!TOKEN:context.abort(grpc.StatusCode.UNAUTHENTICATED,請(qǐng)先登錄)returnhandler.unary_unary(request,context)# Token 正確繼續(xù)執(zhí)行原來(lái)的業(yè)務(wù)函數(shù)# 用權(quán)限檢查函數(shù)替換原來(lái)的業(yè)務(wù)處理函數(shù)。returnhandler._replace(unary_unarycheck_token)defserver(address[::]:50051,max_workers2):啟動(dòng) gRPC 服務(wù)。# 創(chuàng)建服務(wù)并安裝 AuthInterceptor 權(quán)限攔截器grpc_servergrpc.server(futures.ThreadPoolExecutor(max_workersmax_workers),interceptors[AuthInterceptor()])# 注冊(cè)登錄接口grpc_server.add_generic_rpc_handlers((grpc.method_handlers_generic_handler(auth.Auth,{Login:json_handler(login)},),# 注冊(cè)需要登錄的問候接口。grpc.method_handlers_generic_handler(hello.Greeter,{SayHello:json_handler(say_hello)},),))grpc_server.add_insecure_port(address)grpc_server.start()print(fgRPC 服務(wù)已啟動(dòng)監(jiān)聽地址{address})grpc_server.wait_for_termination()if__name____main__:server()客戶端代碼importjsonimportgrpc# 連接地址必須和 server.py 中的端口一致。withgrpc.insecure_channel(127.0.0.1:50051)aschannel:# 第一步調(diào)用登錄接口獲取服務(wù)端保存的 Token。loginchannel.unary_unary(/auth.Auth/Login,request_serializerlambdadata:json.dumps(data).encode(utf-8),response_deserializerlambdadata:json.loads(data.decode(utf-8)),)login_resultlogin({username:admin,password:123456})print(login_result[message])# 第二步調(diào)用需要登錄的 SayHello 接口。say_hellochannel.unary_unary(/hello.Greeter/SayHello,request_serializerlambdadata:json.dumps(data).encode(utf-8),response_deserializerlambdadata:json.loads(data.decode(utf-8)),)# 登錄成功后把 Token 和名稱一起發(fā)送給服務(wù)端。responsesay_hello({name:World,token:login_result[token]})print(response[message])# 第三步不發(fā)送 Token模擬未登錄調(diào)用。try:say_hello({name:World})exceptgrpc.RpcErroraserror:print(f未登錄{error.details()})