跳转至

saya

Saya 相关的工具

decorate 🔗

decorate(*args) -> Wrapper

给指定参数名称附加装饰器

Parameters:

  • name (str | Dict[str, Decorator]) –

    参数名称或与装饰器的映射

  • decorator (Decorator) –

    装饰器

Returns:

  • Wrapper –

    Callable[[T_Callable], T_Callable]: 装饰器

Source code in src/graia/ariadne/util/saya.py 
 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
121
122
123
124
125
126
def decorate(*args) -> Wrapper:
    """给指定参数名称附加装饰器

    Args:
        name (str | Dict[str, Decorator]): 参数名称或与装饰器的映射
        decorator (Decorator): 装饰器

    Returns:
        Callable[[T_Callable], T_Callable]: 装饰器
    """
    arg: dict[str, Decorator] | list[Decorator]
    if isinstance(args[0], str):
        name: str = args[0]
        decorator: Decorator = args[1]
        arg = {name: decorator}
    elif isinstance(args[0], dict):
        arg = args[0]
    else:
        arg = list(args)

    def wrapper(func: T_Callable) -> T_Callable:
        cube = ensure_cube_as_listener(func)
        if isinstance(arg, list):
            cube.metaclass.decorators.extend(arg)
        elif isinstance(arg, dict):
            sig = inspect.signature(func)
            sig.parameters
            for param in sig.parameters.values():
                if param.name in arg:
                    setattr(param, "_default", arg[param.name])
            func.__signature__ = sig
        return func

    return wrapper

dispatch 🔗

dispatch(*dispatcher: T_Dispatcher) -> Wrapper

附加参数解析器，最后必须接 listen 才能起效

Parameters:

  • *dispatcher (T_Dispatcher) –

    参数解析器

Returns:

  • Wrapper –

    Callable[[T_Callable], T_Callable]: 装饰器

Source code in src/graia/ariadne/util/saya.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def dispatch(*dispatcher: T_Dispatcher) -> Wrapper:
    """附加参数解析器，最后必须接 `listen` 才能起效

    Args:
        *dispatcher (T_Dispatcher): 参数解析器

    Returns:
        Callable[[T_Callable], T_Callable]: 装饰器
    """

    def wrapper(func: T_Callable) -> T_Callable:
        cube: Cube[ListenerSchema] = ensure_cube_as_listener(func)
        cube.metaclass.inline_dispatchers.extend(dispatcher)
        return func

    return wrapper

listen 🔗

listen(*event: type[Dispatchable] | str) -> Wrapper

在当前 Saya Channel 中监听指定事件

Parameters:

  • *event (Union[Type[Dispatchable], str]) –

    事件类型或事件名称

Returns:

  • Wrapper –

    Callable[[T_Callable], T_Callable]: 装饰器

Source code in src/graia/ariadne/util/saya.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def listen(*event: type[Dispatchable] | str) -> Wrapper:
    """在当前 Saya Channel 中监听指定事件

    Args:
        *event (Union[Type[Dispatchable], str]): 事件类型或事件名称

    Returns:
        Callable[[T_Callable], T_Callable]: 装饰器
    """
    EVENTS: dict[str, type[Dispatchable]] = {e.__name__: e for e in gen_subclass(Dispatchable)}
    events: list[type[Dispatchable]] = [e if isinstance(e, type) else EVENTS[e] for e in event]

    def wrapper(func: T_Callable) -> T_Callable:
        cube = ensure_cube_as_listener(func)
        cube.metaclass.listening_events.extend(events)
        return func

    return wrapper

priority 🔗

priority(priority: int, *events: type[Dispatchable]) -> Wrapper

设置事件优先级

Parameters:

  • priority (int) –

    事件优先级

  • *events (Type[Dispatchable]) –

    提供时则会设置这些事件的优先级, 否则设置全局优先级

Returns:

  • Wrapper –

    Callable[[T_Callable], T_Callable]: 装饰器

Source code in src/graia/ariadne/util/saya.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def priority(priority: int, *events: type[Dispatchable]) -> Wrapper:
    """设置事件优先级

    Args:
        priority (int): 事件优先级
        *events (Type[Dispatchable]): 提供时则会设置这些事件的优先级, 否则设置全局优先级

    Returns:
        Callable[[T_Callable], T_Callable]: 装饰器
    """

    def wrapper(func: T_Callable) -> T_Callable:
        cube = ensure_cube_as_listener(func)
        cube.metaclass.priority = priority
        if events:
            extra: dict[type[Dispatchable] | None, int] = getattr(cube.metaclass, "extra_priorities", {})
            extra.update((e, priority) for e in events)
        return func

    return wrapper