o NK&h@sddlmZddlmZddlmZddlmZmZm Z m Z ddl m Z ddl mZddlmZddlmZmZdd lmZGd d d e d Zd S)) annotations) Coroutine)Enum)AnyCallableOptionalUnion) SanicMeta) FutureSignal) SignalHandler)EventSignal) HashableDictc@sZeZdZd%ddZd&dd Zd d d d d d'ddZ d(d)ddZd*d d!Zd+d#d$Zd S), SignalMixinreturnNonecOs t|_dSN)set_future_signals)selfargskwargsrG/var/www/html/venv/lib/python3.10/site-packages/sanic/mixins/signals.py__init__s zSignalMixin.__init__signalr r cCtrNotImplementedError)rrrrr _apply_signalzSignalMixin._apply_signalTNr)apply condition exclusivepriorityeventUnion[str, Enum]r!boolr"Optional[dict[str, Any]]r#r$int(Callable[[SignalHandler], SignalHandler]cs4t|tr t|jn|dfdd }|S)a For creating a signal handler, used similar to a route handler: .. code-block:: python @app.signal("foo.bar.") async def signal_handler(thing, **kwargs): print(f"[signal_handler] {thing=}", kwargs) :param event: Representation of the event in ``one.two.three`` form :type event: str :param apply: For lazy evaluation, defaults to ``True`` :type apply: bool, optional :param condition: For use with the ``condition`` argument in dispatch filtering, defaults to ``None`` :param exclusive: When ``True``, the signal can only be dispatched when the condition has been met. When ``False``, the signal can be dispatched either with or without it. *THIS IS INAPPLICABLE TO BLUEPRINT SIGNALS. THEY ARE ALWAYS NON-EXCLUSIVE*, defaults to ``True`` :type condition: Dict[str, Any], optional handlerr cs6t|tpi}j|r||Sr)r rraddr)r+ future_signalr!r" event_valuer#r$rrr decorator6s   z%SignalMixin.signal..decoratorN)r+r ) isinstancerstrvalue)rr%r!r"r#r$r0rr.rrszSignalMixin.signalr+Optional[Callable[..., Any]]Callable[..., Any]cCs(|sdd}|}|j|||d||S)a9Registers a signal handler for a specific event. Args: handler (Optional[Callable[..., Any]]): The function to be called when the event occurs. Defaults to a noop if not provided. event (str): The name of the event to listen for. condition (Optional[Dict[str, Any]]): Optional condition to filter the event triggering. Defaults to `None`. exclusive (bool): Whether or not the handler is exclusive. When `True`, the signal can only be dispatched when the `condition` has been met. *This is inapplicable to blueprint signals, which are **ALWAYS** non-exclusive.* Defaults to `True`. Returns: Callable[..., Any]: The handler that was registered. csdSrr)contextrrrnoopasz$SignalMixin.add_signal..noop)r%r"r#)r)rr+r%r"r#r7rrr add_signalGszSignalMixin.add_signalr2cCrrr)rr%rrrr%ir zSignalMixin.event=Callable[[SignalMixin, Exception], Coroutine[Any, Any, None]]cs$dfdd }tj|dS)aRegister an exception handler for logging or processing. This method allows the registration of a custom exception handler to catch and process exceptions that occur in the application. Unlike a typical exception handler that might modify the response to the client, this is intended to capture exceptions for logging or other internal processing, such as sending them to an error reporting utility. Args: handler (Callable): A coroutine function that takes the application instance and the exception as arguments. It will be called when an exception occurs within the application's lifecycle. Example: ```python app = Sanic("TestApp") @app.catch_exception async def report_exception(app: Sanic, exception: Exception): logging.error(f"An exception occurred: {exception}") # Send to an error reporting service await error_service.report(exception) # Any unhandled exceptions within the application will now be # logged and reported to the error service. ``` exception Exceptioncs|IdHdSrr)r:r+rrrsignal_handlersz3SignalMixin.catch_exception..signal_handlerN)r:r;)rr SERVER_EXCEPTION_REPORT)rr+r=rr<rcatch_exceptionls!zSignalMixin.catch_exception)rr)rr rr ) r%r&r!r'r"r(r#r'r$r)rr*)NT) r+r4r%r&r"r(r#r'rr5)r%r2)r+r9rr) __name__ __module__ __qualname__rrrr8r%r?rrrrrs  6 "r) metaclassN) __future__rcollections.abcrenumrtypingrrrrsanic.base.metar sanic.models.futuresr sanic.models.handler_typesr sanic.signalsr r sanic.typesrrrrrrs