53 lines
1.5 KiB
Python
53 lines
1.5 KiB
Python
from typing import List
|
|
|
|
|
|
class GeneralInterceptor():
|
|
def __init__(
|
|
self,
|
|
module: str,
|
|
before_list=None,
|
|
after_list=None,
|
|
before_insert=None,
|
|
after_insert=None,
|
|
before_update=None,
|
|
after_update=None,
|
|
before_batch_insert=None,
|
|
after_batch_insert=None,
|
|
before_batch_update=None,
|
|
after_batch_update=None,
|
|
before_delete=None,
|
|
after_delete=None,
|
|
):
|
|
self.module = module
|
|
self.before_list = before_list
|
|
self.after_list = after_list
|
|
self.before_insert = before_insert
|
|
self.after_insert = after_insert
|
|
self.before_update = before_update
|
|
self.after_update = after_update
|
|
self.before_batch_insert = before_batch_insert
|
|
self.after_batch_insert = after_batch_insert
|
|
self.before_batch_update = before_batch_update
|
|
self.after_batch_update = after_batch_update
|
|
self.before_delete = before_delete
|
|
self.after_delete = after_delete
|
|
|
|
|
|
_general_interceptors: List[GeneralInterceptor] = []
|
|
|
|
|
|
def add_general_interceptor(interceptor: GeneralInterceptor):
|
|
_general_interceptors.append(interceptor)
|
|
|
|
|
|
async def invoke_interceptor(method: str, module: str, *args):
|
|
if module.startswith('/'):
|
|
module = module[1:]
|
|
if module.endswith('/'):
|
|
module = module[:-1]
|
|
match_interceptors = [item for item in _general_interceptors if item.module == module]
|
|
for item in match_interceptors:
|
|
interceptor_method = getattr(item, method, None) # 使用 getattr 获取方法
|
|
if interceptor_method and callable(interceptor_method):
|
|
await interceptor_method(*args) # 调用方法
|