跳转至

interrupt

Broadcast Interrupt 相关的工具

AnnotationWaiter 🔗

Bases: _ExtendedWaiter[T, T_E]

用于直接获取对应标注的 Waiter.

Source code in src/graia/ariadne/util/interrupt.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
class AnnotationWaiter(_ExtendedWaiter[T, T_E]):
    """用于直接获取对应标注的 Waiter."""

    def __init__(
        self,
        annotation: Type[T],
        events: List[Type[T_E]],
        dispatchers: Optional[List[T_Dispatcher]] = None,
        decorator: Optional[Decorator] = None,
        headless_decorators: Optional[List[Decorator]] = None,
        extra_validator: Optional[Callable[[T_E], bool]] = None,
        priority: int = 15,
        block_propagation: bool = False,
    ) -> None:
        """
        Args:
            annotation (Type[T]): 参数标注
            events (List[Type[T_E]]): 事件类型
            dispatchers (Optional[List[T_Dispatcher]], optional): Dispatcher 列表
            decorator (Decorator, optional): 可选的参数装饰器
            headless_decorators (Optional[List[Decorator]], optional): 无头 Decorator 列表
            extra_validator (Optional[Callable[[T_E], bool]], optional): 额外的验证器
            priority (int, optional): 优先级, 越小越靠前
            block_propagation (bool): 是否阻止事件往下传播
        """
        super().__init__(events, dispatchers or [], headless_decorators or [], priority, block_propagation)
        self.annotation: Type[T] = annotation
        self.decorator = decorator
        self.extra_validator = extra_validator

    async def detected_event(self, dii: DispatcherInterface) -> T:
        event = cast(T_E, dii.event)
        if self.extra_validator and not self.extra_validator(event):
            raise ExecutionStop
        return await dii.lookup_param("__AnnotationWaiter_annotation__", self.annotation, self.decorator)

__init__ 🔗

__init__(
    annotation: Type[T],
    events: List[Type[T_E]],
    dispatchers: Optional[List[T_Dispatcher]] = None,
    decorator: Optional[Decorator] = None,
    headless_decorators: Optional[List[Decorator]] = None,
    extra_validator: Optional[Callable[[T_E], bool]] = None,
    priority: int = 15,
    block_propagation: bool = False,
) -> None

Parameters:

  • annotation (Type[T]) –

    参数标注

  • events (List[Type[T_E]]) –

    事件类型

  • dispatchers (Optional[List[T_Dispatcher]]) –

    Dispatcher 列表

  • decorator (Decorator) –

    可选的参数装饰器

  • headless_decorators (Optional[List[Decorator]]) –

    无头 Decorator 列表

  • extra_validator (Optional[Callable[[T_E], bool]]) –

    额外的验证器

  • priority (int) –

    优先级, 越小越靠前

  • block_propagation (bool) –

    是否阻止事件往下传播

Source code in src/graia/ariadne/util/interrupt.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def __init__(
    self,
    annotation: Type[T],
    events: List[Type[T_E]],
    dispatchers: Optional[List[T_Dispatcher]] = None,
    decorator: Optional[Decorator] = None,
    headless_decorators: Optional[List[Decorator]] = None,
    extra_validator: Optional[Callable[[T_E], bool]] = None,
    priority: int = 15,
    block_propagation: bool = False,
) -> None:
    """
    Args:
        annotation (Type[T]): 参数标注
        events (List[Type[T_E]]): 事件类型
        dispatchers (Optional[List[T_Dispatcher]], optional): Dispatcher 列表
        decorator (Decorator, optional): 可选的参数装饰器
        headless_decorators (Optional[List[Decorator]], optional): 无头 Decorator 列表
        extra_validator (Optional[Callable[[T_E], bool]], optional): 额外的验证器
        priority (int, optional): 优先级, 越小越靠前
        block_propagation (bool): 是否阻止事件往下传播
    """
    super().__init__(events, dispatchers or [], headless_decorators or [], priority, block_propagation)
    self.annotation: Type[T] = annotation
    self.decorator = decorator
    self.extra_validator = extra_validator

EventWaiter 🔗

Bases: _ExtendedWaiter[T_E, T_E]

将 Waiter.create_using_event 封装了一层.

Source code in src/graia/ariadne/util/interrupt.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
class EventWaiter(_ExtendedWaiter[T_E, T_E]):
    """将 Waiter.create_using_event 封装了一层."""

    def __init__(
        self,
        events: List[Type[T_E]],
        dispatchers: Optional[List[T_Dispatcher]] = None,
        decorators: Optional[List[Decorator]] = None,
        extra_validator: Optional[Callable[[T_E], bool]] = None,
        priority: int = 15,
        block_propagation: bool = False,
    ) -> None:
        """
        Args:
            events (List[Type[T_E]]): 事件类型
            dispatchers (Optional[List[T_Dispatcher]], optional): Dispatcher 列表
            decorators (Optional[List[Decorator]], optional): Decorator 列表
            extra_validator (Optional[Callable[[T_E], bool]], optional): 额外的验证器
            priority (int, optional): 优先级, 越小越靠前
            block_propagation (bool): 是否阻止事件往下传播
        """
        super().__init__(events, dispatchers or [], decorators or [], priority, block_propagation)
        self.extra_validator = extra_validator

    async def detected_event(self, ev: Dispatchable) -> T_E:
        event = cast(T_E, ev)
        if self.extra_validator and not self.extra_validator(event):
            raise ExecutionStop
        return event

__init__ 🔗

__init__(
    events: List[Type[T_E]],
    dispatchers: Optional[List[T_Dispatcher]] = None,
    decorators: Optional[List[Decorator]] = None,
    extra_validator: Optional[Callable[[T_E], bool]] = None,
    priority: int = 15,
    block_propagation: bool = False,
) -> None

Parameters:

  • events (List[Type[T_E]]) –

    事件类型

  • dispatchers (Optional[List[T_Dispatcher]]) –

    Dispatcher 列表

  • decorators (Optional[List[Decorator]]) –

    Decorator 列表

  • extra_validator (Optional[Callable[[T_E], bool]]) –

    额外的验证器

  • priority (int) –

    优先级, 越小越靠前

  • block_propagation (bool) –

    是否阻止事件往下传播

Source code in src/graia/ariadne/util/interrupt.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def __init__(
    self,
    events: List[Type[T_E]],
    dispatchers: Optional[List[T_Dispatcher]] = None,
    decorators: Optional[List[Decorator]] = None,
    extra_validator: Optional[Callable[[T_E], bool]] = None,
    priority: int = 15,
    block_propagation: bool = False,
) -> None:
    """
    Args:
        events (List[Type[T_E]]): 事件类型
        dispatchers (Optional[List[T_Dispatcher]], optional): Dispatcher 列表
        decorators (Optional[List[Decorator]], optional): Decorator 列表
        extra_validator (Optional[Callable[[T_E], bool]], optional): 额外的验证器
        priority (int, optional): 优先级, 越小越靠前
        block_propagation (bool): 是否阻止事件往下传播
    """
    super().__init__(events, dispatchers or [], decorators or [], priority, block_propagation)
    self.extra_validator = extra_validator

FunctionWaiter 🔗

Bases: _ExtendedWaiter[T, Dispatchable]

将 Waiter.create_using_function 封装了一层

Source code in src/graia/ariadne/util/interrupt.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
class FunctionWaiter(_ExtendedWaiter[T, Dispatchable]):
    """将 Waiter.create_using_function 封装了一层"""

    def __init__(
        self,
        func: Callable[..., Awaitable[Optional[T]]],
        events: List[Type[Dispatchable]],
        dispatchers: Optional[List[T_Dispatcher]] = None,
        decorators: Optional[List[Decorator]] = None,
        priority: int = 15,
        block_propagation: bool = False,
    ) -> None:
        """
        Args:
            func (Callable): 调用函数
            events (List[Type[Dispatchable]]): 事件类型
            dispatchers (Optional[List[T_Dispatcher]]): 广播器
            decorators (Optional[List[Decorator]]): 装饰器
            priority (int): 优先级
            block_propagation (bool): 是否阻止事件往下传播
        """
        super().__init__(events, dispatchers or [], decorators or [], priority, block_propagation)
        self.detected_event = func  # type: ignore

__init__ 🔗

__init__(
    func: Callable[..., Awaitable[Optional[T]]],
    events: List[Type[Dispatchable]],
    dispatchers: Optional[List[T_Dispatcher]] = None,
    decorators: Optional[List[Decorator]] = None,
    priority: int = 15,
    block_propagation: bool = False,
) -> None

Parameters:

  • func (Callable) –

    调用函数

  • events (List[Type[Dispatchable]]) –

    事件类型

  • dispatchers (Optional[List[T_Dispatcher]]) –

    广播器

  • decorators (Optional[List[Decorator]]) –

    装饰器

  • priority (int) –

    优先级

  • block_propagation (bool) –

    是否阻止事件往下传播

Source code in src/graia/ariadne/util/interrupt.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def __init__(
    self,
    func: Callable[..., Awaitable[Optional[T]]],
    events: List[Type[Dispatchable]],
    dispatchers: Optional[List[T_Dispatcher]] = None,
    decorators: Optional[List[Decorator]] = None,
    priority: int = 15,
    block_propagation: bool = False,
) -> None:
    """
    Args:
        func (Callable): 调用函数
        events (List[Type[Dispatchable]]): 事件类型
        dispatchers (Optional[List[T_Dispatcher]]): 广播器
        decorators (Optional[List[Decorator]]): 装饰器
        priority (int): 优先级
        block_propagation (bool): 是否阻止事件往下传播
    """
    super().__init__(events, dispatchers or [], decorators or [], priority, block_propagation)
    self.detected_event = func  # type: ignore