尧图网站建设 
在gRPC的权限校验中写一个类然后继承grpc.ServerInterceptor即可使用权限校验每隔函数在调用前都会调用这个函数这个函数可用于权限校验或是一些数据准备等操作。文章目录权限校验示例程序服务端代码客户端代码带Login的权限管理服务端代码客户端代码权限校验示例程序服务端代码importjsonfromconcurrentimportfuturesimportgrpc# 业务函数收到客户端请求后返回问候消息。defsay_hello(request,_context):return{message:fHello,{request[name]}!}# 拦截器是 gRPC 的固定扩展写法可以在请求到达业务函数前做日志、鉴权等处理。# 这里为了保持简单只打印请求方法不做真正的权限校验。classLogInterceptor(grpc.ServerInterceptor):defintercept_service(self,next_step,handler_call_details):print(f收到请求{handler_call_details.method})returnnext_step(handler_call_details)defserver(envdev,address[::]:50051,max_workers2):启动 gRPC 服务。 env运行环境名称当前只用于打印提示。 address监听地址格式通常是“主机:端口”。 max_workers线程池中的最大工作线程数。 print(f当前运行环境{env})# 这是 gRPC 创建服务端的固定写法。grpc_servergrpc.server(futures.ThreadPoolExecutor(max_workersmax_workers),interceptors[LogInterceptor()])# 下面是本示例的 JSON 请求处理器。# 标准 gRPC 项目通常会用 .proto 文件自动生成这部分代码。handlergrpc.unary_unary_rpc_method_handler(say_hello,request_deserializerlambdadata:json.loads(data.decode(utf-8)),response_serializerlambdaresponse:json.dumps(response).encode(utf-8),)# 注册服务和方法这是手动注册 RPC 方法的写法。grpc_server.add_generic_rpc_handlers((grpc.method_handlers_generic_handler(hello.Greeter,{SayHello:handler}),))# 添加监听端口、启动服务、持续等待是 gRPC 服务启动的固定流程。grpc_server.add_insecure_port(address)grpc_server.start()print(fgRPC 服务已启动监听地址{address})grpc_server.wait_for_termination()# 程序从这里开始执行调用 server() 启动服务。if__name____main__:server()客户端代码importjsonimportgrpc# 连接地址必须和 server.py 中的端口一致。withgrpc.insecure_channel(127.0.0.1:50051)aschannel:# 指定要调用的服务方法/服务名/方法名。say_hellochannel.unary_unary(/hello.Greeter/SayHello,# 发送请求前Python 字典转换为 JSON 字节。request_serializerlambdadata:json.dumps(data).encode(utf-8),# 收到响应后JSON 字节转换为 Python 字典。response_deserializerlambdadata:json.loads(data.decode(utf-8)),)# 发送名称并获取服务端返回的问候消息。responsesay_hello({name:World})print(response[message])带Login的权限管理使用带login的权限管理核心在于AuthInterceptor这个类中当method /auth.Auth/Login时直接执行校验当执行其他的method的时候直接替换调用check_token这个函数校验token如果可以就通过handler.unary_unary(request, context)继续执行如果不行就通过context.abort(grpc.StatusCode.UNAUTHENTICATED, 请先登录)直接返回不进行后续的执行服务端代码importjsonfromconcurrentimportfuturesimportgrpc# 服务端保存登录成功后的 Token。TOKENNone# 登录接口不需要 Token登录成功后生成并保存 Token。deflogin(request,_context):globalTOKENifrequest[username]!adminorrequest[password]!123456:return{success:False,message:用户名或密码错误}TOKENtoken-123return{success:True,token:TOKEN,message:登录成功}# 只有通过 Token 校验后才会执行这个业务函数。defsay_hello(request,_context):return{message:fHello,{request[name]}!}# 将普通 Python 函数注册成 gRPC 方法并处理 JSON 转换。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),)# 在业务函数执行前检查 Token。classAuthInterceptor(grpc.ServerInterceptor):defintercept_service(self,continuation,handler_call_details):methodhandler_call_details.method handlercontinuation(handler_call_details)ifhandlerisNone:returnNone# 登录接口不能检查 Token否则客户端无法登录。ifmethod/auth.Auth/Login:returnhandler# 其他接口都必须携带正确的 Token。defcheck_token(request,context):ifrequest.get(token)!TOKEN:context.abort(grpc.StatusCode.UNAUTHENTICATED,请先登录)returnhandler.unary_unary(request,context)# Token 正确继续执行原来的业务函数# 用权限检查函数替换原来的业务处理函数。returnhandler._replace(unary_unarycheck_token)defserver(address[::]:50051,max_workers2):启动 gRPC 服务。# 创建服务并安装 AuthInterceptor 权限拦截器grpc_servergrpc.server(futures.ThreadPoolExecutor(max_workersmax_workers),interceptors[AuthInterceptor()])# 注册登录接口grpc_server.add_generic_rpc_handlers((grpc.method_handlers_generic_handler(auth.Auth,{Login:json_handler(login)},),# 注册需要登录的问候接口。grpc.method_handlers_generic_handler(hello.Greeter,{SayHello:json_handler(say_hello)},),))grpc_server.add_insecure_port(address)grpc_server.start()print(fgRPC 服务已启动监听地址{address})grpc_server.wait_for_termination()if__name____main__:server()客户端代码importjsonimportgrpc# 连接地址必须和 server.py 中的端口一致。withgrpc.insecure_channel(127.0.0.1:50051)aschannel:# 第一步调用登录接口获取服务端保存的 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])# 第二步调用需要登录的 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 和名称一起发送给服务端。responsesay_hello({name:World,token:login_result[token]})print(response[message])# 第三步不发送 Token模拟未登录调用。try:say_hello({name:World})exceptgrpc.RpcErroraserror:print(f未登录{error.details()}) gRPC微服务搭建（学习阶段2:权限校验）
在gRPC的权限校验中写一个类然后继承grpc.ServerInterceptor即可使用权限校验每隔函数在调用前都会调用这个函数这个函数可用于权限校验或是一些数据准备等操作。文章目录权限校验示例程序服务端代码客户端代码带Login的权限管理服务端代码客户端代码权限校验示例程序服务端代码importjsonfromconcurrentimportfuturesimportgrpc# 业务函数收到客户端请求后返回问候消息。defsay_hello(request,_context):return{message:fHello,{request[name]}!}# 拦截器是 gRPC 的固定扩展写法可以在请求到达业务函数前做日志、鉴权等处理。# 这里为了保持简单只打印请求方法不做真正的权限校验。classLogInterceptor(grpc.ServerInterceptor):defintercept_service(self,next_step,handler_call_details):print(f收到请求{handler_call_details.method})returnnext_step(handler_call_details)defserver(envdev,address[::]:50051,max_workers2):启动 gRPC 服务。 env运行环境名称当前只用于打印提示。 address监听地址格式通常是“主机:端口”。 max_workers线程池中的最大工作线程数。 print(f当前运行环境{env})# 这是 gRPC 创建服务端的固定写法。grpc_servergrpc.server(futures.ThreadPoolExecutor(max_workersmax_workers),interceptors[LogInterceptor()])# 下面是本示例的 JSON 请求处理器。# 标准 gRPC 项目通常会用 .proto 文件自动生成这部分代码。handlergrpc.unary_unary_rpc_method_handler(say_hello,request_deserializerlambdadata:json.loads(data.decode(utf-8)),response_serializerlambdaresponse:json.dumps(response).encode(utf-8),)# 注册服务和方法这是手动注册 RPC 方法的写法。grpc_server.add_generic_rpc_handlers((grpc.method_handlers_generic_handler(hello.Greeter,{SayHello:handler}),))# 添加监听端口、启动服务、持续等待是 gRPC 服务启动的固定流程。grpc_server.add_insecure_port(address)grpc_server.start()print(fgRPC 服务已启动监听地址{address})grpc_server.wait_for_termination()# 程序从这里开始执行调用 server() 启动服务。if__name____main__:server()客户端代码importjsonimportgrpc# 连接地址必须和 server.py 中的端口一致。withgrpc.insecure_channel(127.0.0.1:50051)aschannel:# 指定要调用的服务方法/服务名/方法名。say_hellochannel.unary_unary(/hello.Greeter/SayHello,# 发送请求前Python 字典转换为 JSON 字节。request_serializerlambdadata:json.dumps(data).encode(utf-8),# 收到响应后JSON 字节转换为 Python 字典。response_deserializerlambdadata:json.loads(data.decode(utf-8)),)# 发送名称并获取服务端返回的问候消息。responsesay_hello({name:World})print(response[message])带Login的权限管理使用带login的权限管理核心在于AuthInterceptor这个类中当method /auth.Auth/Login时直接执行校验当执行其他的method的时候直接替换调用check_token这个函数校验token如果可以就通过handler.unary_unary(request, context)继续执行如果不行就通过context.abort(grpc.StatusCode.UNAUTHENTICATED, 请先登录)直接返回不进行后续的执行服务端代码importjsonfromconcurrentimportfuturesimportgrpc# 服务端保存登录成功后的 Token。TOKENNone# 登录接口不需要 Token登录成功后生成并保存 Token。deflogin(request,_context):globalTOKENifrequest[username]!adminorrequest[password]!123456:return{success:False,message:用户名或密码错误}TOKENtoken-123return{success:True,token:TOKEN,message:登录成功}# 只有通过 Token 校验后才会执行这个业务函数。defsay_hello(request,_context):return{message:fHello,{request[name]}!}# 将普通 Python 函数注册成 gRPC 方法并处理 JSON 转换。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),)# 在业务函数执行前检查 Token。classAuthInterceptor(grpc.ServerInterceptor):defintercept_service(self,continuation,handler_call_details):methodhandler_call_details.method handlercontinuation(handler_call_details)ifhandlerisNone:returnNone# 登录接口不能检查 Token否则客户端无法登录。ifmethod/auth.Auth/Login:returnhandler# 其他接口都必须携带正确的 Token。defcheck_token(request,context):ifrequest.get(token)!TOKEN:context.abort(grpc.StatusCode.UNAUTHENTICATED,请先登录)returnhandler.unary_unary(request,context)# Token 正确继续执行原来的业务函数# 用权限检查函数替换原来的业务处理函数。returnhandler._replace(unary_unarycheck_token)defserver(address[::]:50051,max_workers2):启动 gRPC 服务。# 创建服务并安装 AuthInterceptor 权限拦截器grpc_servergrpc.server(futures.ThreadPoolExecutor(max_workersmax_workers),interceptors[AuthInterceptor()])# 注册登录接口grpc_server.add_generic_rpc_handlers((grpc.method_handlers_generic_handler(auth.Auth,{Login:json_handler(login)},),# 注册需要登录的问候接口。grpc.method_handlers_generic_handler(hello.Greeter,{SayHello:json_handler(say_hello)},),))grpc_server.add_insecure_port(address)grpc_server.start()print(fgRPC 服务已启动监听地址{address})grpc_server.wait_for_termination()if__name____main__:server()客户端代码importjsonimportgrpc# 连接地址必须和 server.py 中的端口一致。withgrpc.insecure_channel(127.0.0.1:50051)aschannel:# 第一步调用登录接口获取服务端保存的 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])# 第二步调用需要登录的 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 和名称一起发送给服务端。responsesay_hello({name:World,token:login_result[token]})print(response[message])# 第三步不发送 Token模拟未登录调用。try:say_hello({name:World})exceptgrpc.RpcErroraserror:print(f未登录{error.details()})