preface

In view of the recent things at hand a little busy, so stop more, in order to live up to expectations, or need to continue to write some things before, even to their own account.

In the next installment, we focused on batch registration of services, and this installment referred to defining our FastAPi-like events and middleware additions:

Example for adding events:

@app.on_event('startup') def startup(): Print ("startup=========1 ") sync_redis_client.init_app(app) @app.on_event('shutdown') def shutdown(): Def ceshiguanbi(): print("shutdown========= ") Add_event_handler (event_type='shutdown', func=ceshiguanbi) # Add service handlerCopy the code

Examples of middleware additions:

Add_middleware class ServiceLogMiddleware2222(BaseMiddleware): def before_handler(self, request: Any, context: grpc.ServicerContext, method_name): pass print(" I am a custom middleware message ") def after_handler(self, context: Grpc. ServicerContext, response): def error_grpc_exc_handler(self, error): Pass mode 2: # # add service processing app. Add_servicer (Greeter) from grpcframe. Middlewares. Tracer. The middleware import Zyxopentracing app.add_middleware(Zyxopentracing) from example.hello_srv.middlewares.authinterceptor import AuthMiddleware # PS: Middleware registration, last to register, first to execute, following the Onion model (need to be based on Bsea and other Middleware, native will execute first!) app.add_middleware(AuthMiddleware)Copy the code

Event Registry

Event Registration Mode

In view of the fact that the framework structure was not very good before, some new adjustments were made as follows:

An example of the relevant time registration is shown in the code:

from grpcframe.zyxgrpc.base_app import FastGrpcApp from grpcframe.exts.redis.client import sync_redis_client from grpcframe.exts.register.consul import ConsulRegister from grpcframe.zyxgrpc.base_middleware import BaseMiddleware,Any Host ='127.0.0.1',port=8500, Server_name ='FastGrpcApp',server_id="23423424234") # instantiate a server with a registry app = FastGrpcApp(port=45564,max_workers=20,is_certificate=False,register_center=None,debug=True) @app.on_event('startup') def Startup (): print("startup=========1 ") sync_redis_client.init_app(app) @app.on_event('shutdown') def shutdown(): Print (" shutdown = = = = = = = = = "") if __name__ = =" __main__ ': # separate loading app. Run_serve ()Copy the code

Encapsulate ideas and steps

Here the main basic reference fastAPI I design, very simple and not complex.

Step 1: Define the list of stored registry functions

Self.on_startup = [] if on_startup is None else list(on_startup) self.on_shutdown = [] if on_shutdown is None  else list(on_shutdown)Copy the code

Step 2: Define add event or decorator functions

Def on_event(self, event_type: STR) -> typing.Callable: def decorator(func: typing.Callable) -> typing.Callable: Self. Add_event_handler (event_type, func) return func return decorator # def add_event_handler(self, event_type, func) event_type: str, func: typing.Callable) -> None: Asser_type in ("startup", "shutdown") if event_type == "startup": self.on_startup.append(func) else: self.on_shutdown.append(func)Copy the code

Step 3: Define the event handler

# = = = = = = = = = = = = = = = event processing module mechanism = = = = = = = = = = = = = = = def startup (self) - > None: for handler in the self. The on_startup: handler() def shutdown(self) -> None: for handler in self.on_shutdown: handler()Copy the code

Step 4: Call the event handler in the appropriate place

Summary: The summary is to use a monument to store a bunch of functions and make function traversal calls when appropriate.

Middleware article

Examples of middleware definitions:

from grpcframe.zyxgrpc.base_middleware import BaseMiddlewareGrequestcontext from typing import Any from grpc_interceptor.exceptions import GrpcException import grpc from grpcframe.exts.grpclocal.request import Request # In order to to the types of tips from grpcframe. Zyxgrpc. Globals import grequestcontext grequestcontext: Request = grequestcontext # @app.add_middleware class AuthMiddleware(BaseMiddlewareGrequestcontext): def before_handler(self): Print (" Did I print ---- uninitialized?" , grequestContext.parse_request_metadata) print(" I typed -- no initialization? Def after_handler(self): pass def error_grpc_exc_handler(self,e:GrpcException): pass def finally_handler(self): passCopy the code

Currently, there are several basic inheritance methods supported in the middle, and inheritance extensions need to be defined to implement:

The main models are:

from grpc_interceptor import ServerInterceptor from grpc_interceptor.exceptions import GrpcException import abc import grpc from typing import Any, Callable from grpcframe.zyxgrpc.com pose_app import ComposeApp error handling class # extensions need ErrorBase (metaclass = ABC. ABCMeta) : @abc.abstractmethod def error_grpc_exc_handler(self, exception: GrpcException): Raise NotImplementedError() # raise class FinallyBase(metaclass= abc.abcMeta): @abc.abstractmethod def finally_handler(self): raise NotImplementedError() class BeforeAfterBase(metaclass=abc.ABCMeta): @abc.abstractmethod def before_handler(self, request: Any, context: grpc.ServicerContext, method_name): raise NotImplementedError() @abc.abstractmethod def after_handler(self, context: grpc.ServicerContext, response): raise NotImplementedError() class BeforeAfterBaseNoParameter(metaclass=abc.ABCMeta): @abc.abstractmethod def before_handler(self): raise NotImplementedError() @abc.abstractmethod def after_handler(self): raise NotImplementedError() class BaseMiddlewareNoParameter(grpc.ServerInterceptor, BeforeAfterBaseNoParameter): ServerInterceptor (grPC.serverInterceptor) : if the middleware does not follow the Onion model when it circulates through the middleware, it may not return the corresponding information. # it is always the first to execute ！！！！！！！ ''' def intercept_service(self, continuation, handler_call_details): self.before_handler() resp = continuation(handler_call_details) self.after_handler() return resp # You need to inherit from ServerInterceptor and ABCMeta Class BaseComposeAppMiddleware(ServerInterceptor, ErrorBase, BeforeAfterBase, ComposeApp): def intercept(self, method: Callable, request: Any, context: grpc.ServicerContext, method_name: STR,) -> Any: # callback calls handle self.before_handler(request, context, method_name) Response = method(request, context, method_name) Self.after_handler (context, response) return response except GrpcException as e: self.error_grpc_exc_handler(e) # context.set_code(e.status_code) # context.set_details(e.details) raise finally: Class BaseMiddleware(ServerInterceptor, ErrorBase, BeforeAfterBase): def intercept(self, method: Callable, request: Any, context: grpc.ServicerContext, method_name: str, ) -> Any: Create context object try: # callback calls handle self.before_handler(request, context, method_name) Response = method(request, context, method_name) Self.after_handler (context, response) return response except GrpcException as e: self.error_grpc_exc_handler(e) # context.set_code(e.status_code) # context.set_details(e.details) raise finally: Pass # have the request, the basis of the context callback processing class BaseMiddlewareGrequestcontext (ServerInterceptor ErrorBase, BeforeAfterBaseNoParameter, FinallyBase): def intercept(self, method: Callable, request: Any, context: Grpc. ServicerContext, method_name: STR,) -> Any: # callback calls handle self.before_handler() response = method(request, Self.after_handler () return response except GrpcException as e: self.error_grpc_exc_handler(e) # context.set_code(e.status_code) # context.set_details(e.details) raise finally: self.finally_handler()Copy the code

Example code for middleware registration is as follows:

from grpcframe.zyxgrpc.base_app import FastGrpcApp from grpcframe.exts.redis.client import sync_redis_client from grpcframe.exts.register.consul import ConsulRegister from grpcframe.zyxgrpc.base_middleware import BaseMiddleware,Any Host ='127.0.0.1',port=8500, Server_name ='FastGrpcApp',server_id="23423424234") # instantiate a server with a registry app = FastGrpcApp(port=45564,max_workers=20,is_certificate=False,register_center=None,debug=True) @app.add_middleware class ServiceLogMiddleware2222(BaseMiddleware): def before_handler(self, request: Any, context: Grpc. ServicerContext, method_name): pass print(" I am a custom middleware message -- "1") def after_handler(self, context: Grpc. ServicerContext, response): def error_grpc_exc_handler(self, error): Pass # # add service processing app. Add_servicer (Greeter) from grpcframe. Middlewares. Tracer. The middleware import Zyxopentracing app.add_middleware(Zyxopentracing) from example.hello_srv.middlewares.authinterceptor import AuthMiddleware # PS: Middleware registration, last to register, first to execute, following the Onion model (need to be based on Bsea and other Middleware, native will execute first!) App.add_middleware (AuthMiddleware) if __name__ == '__main__': # Load app.run_serve() aloneCopy the code

Encapsulate ideas and steps

Some middleware prefixes

Some middleware references brokers that use the context of a corresponding global quest, and need to register a global proxy middleware by default.

As for the proxy pattern, introduced in the previous article, here are a few new objects that need to be proxy.

import contextvars import typing from dataclasses import dataclass from grpcframe.utils.singleton import Singleton from grpcframe.exts.grpclocal.local import LocalProxy from functools import partial @dataclass class GManager(metaclass=Singleton): Current_app_request = contextvars.ContextVar('current_app_request', Default =None) # current_app_context = contextvars.ContextVar('current_app_context', Active_tracer_span = contextvars.ContextVar(' current_active_tracer_SPAN ', Default =None) # Current_app_request_context = contextvars.ContextVar(' current_APP_request_context ', default=None) _g = GManager() def set_current_app_request(value): _g.current_app_request.set(value) def get_current_app_request(_g=None): Get () return _G.current_app_request.get () def set_current_app_request_context(value): _g.current_app_request_context.set(value) def get_current_app_request_context(_g=None): Get () return _G.current_app_request_context.get () def get_current_app_request_context2(): Get () return _G.current_app_request_context.get () def set_current_app_context(value): _g.current_app_context.set(value) def get_current_app_context(_g=None): Get () return _G.current_app_context.get () def set_active_tracer_span(value): _g.active_tracer_span.set(value) def get_active_tracer_span(_g=None): Get () return _g.active_tracer_span. Get () # define a current_app_request object in the global proxy g, Grequest = LocalProxy(partial(get_current_app_request, _g)) gcontext = LocalProxy(partial(get_current_app_context, _g)) gactive_tracer_span = LocalProxy(partial(get_active_tracer_span, _g)) grequestcontext = LocalProxy(partial(get_current_app_request_context, _g))Copy the code

Other steps:

Step 1: Define the related middleware base classes for the base

Different base classes implement different ways to implement, see the base class pattern above.

Step 2: Add by adding functions

Using the decorator form is to decorate the class, and you can wrap it yourself if you need to.

from example.hello_srv.middlewares.authinterceptor import AuthMiddleware # PS: Middleware registration, last to register, first to execute, following the Onion model (need to be based on Bsea and other Middleware, native will execute first!) app.add_middleware(AuthMiddleware)Copy the code

The above two points are a brief introduction to the use and ideas of events and middleware modules. Due to the limited time, the description is not detailed enough. If you have any questions, and there should be some problems in the current package, if you have any questions in the use process and are willing to communicate with me, you can contact me at any time. I am also small white, mutual learning learning!

About open source

Fastapi scaffolding encapsulation

Before the first version of FastAPI open source after a scaffolding, some of the contents of the packaging is not very reasonable, but also to some users of the attention, is really flattered,

In fact, after the release of the first version, I was busy with other things, so I shelved it temporarily, and then I rebuilt a version in my spare time.

The following are some of the structures in the latest version, which mainly adds some bulk loading and some common extension packages with the base.

There will be an open source update when things get busy.

GRPC scaffolding encapsulation

The framework basically aims to transform the way of using GRPC into the mode of using FastAPI. However, due to limited skills, some things are still undergoing in-depth transformation, such as the transformation to use route to achieve relevant registration. At present, it is mainly achieved by CBV.

The main function points are:

  • Event registration
  • Batch Registration of services
  • Middleware addition
  • Global context proxy implementation
  • Event processing based on signal library
  • Introduce Pydantic to do simple parameter verification
  • Implementation of default health checks

The project structure is as follows:

Open source address

https://gitee.com/xiaozhong1988/fastgrpc
Copy the code

Pit is unavoidable, hope big guy can give more advice!

conclusion

The above is just a personal combination of their own actual needs, do study practice notes! If there are clerical errors! Welcome criticism and correction! Thank you!

At the end

END

Jane: www.jianshu.com/u/d6960089b…

The Denver nuggets: juejin. Cn/user / 296393…

Public account: wechat search [children to a pot of wolfberry wine tea]

Let students | article | QQ: welcome to learn communication 】 【 308711822