Module graia.broadcast.builtin.factory

Classes

class AsyncDispatcherContextManager (generator_factory: Callable, args=None, kwargs=None)

所有非单函数型 Dispatcher 的基类, 用于为参数解析提供可扩展的支持.

Expand source code

class AsyncDispatcherContextManager(BaseDispatcher):
    generator_factory: Callable[[Any], T_AsyncDispatcherContextManager]
    generator: T_AsyncDispatcherContextManager
    ready: bool = False

    def __init__(self, generator_factory: Callable, args=None, kwargs=None) -> None:
        self.generator_factory = generator_factory
        self.args = args or ()
        self.kwargs = kwargs or {}

    async def beforeExecution(self, interface: "DispatcherInterface"):
        self.generator = self.generator_factory(*self.args, **self.kwargs)
        await self.generator.__anext__()
        await self.generator.asend(interface)
        self.ready = True

    async def catch(self, interface: "DispatcherInterface"):
        if not self.ready:
            return
        status, value = await self.generator.asend((StatusCodeEnum.DISPATCHING, None))
        if status is ResponseCodeEnum.VALUE:
            return value

    async def afterDispatch(
        self,
        interface: "DispatcherInterface",
        exception: Optional[Exception],
        tb: Optional[TracebackType],
    ):
        self.ready = False
        try:
            if not tb:
                await self.generator.asend((StatusCodeEnum.DISPATCH_COMPLETED, None))
            else:
                await self.generator.asend(
                    (StatusCodeEnum.DISPATCH_EXCEPTION, ExcInfo(exception, tb))
                )
        except StopIteration:
            pass

    async def afterExecution(
        self,
        interface: "DispatcherInterface",
        exception: Optional[Exception],
        tb: Optional[TracebackType],
    ):
        try:
            if not tb:
                await self.generator.asend((StatusCodeEnum.EXECUTION_COMPLETED, None))
            else:
                await self.generator.asend(
                    (StatusCodeEnum.EXECUTION_EXCEPTION, ExcInfo(exception, tb))
                )
        except StopIteration:
            pass

Ancestors

• BaseDispatcher

Subclasses

• Kanata

Class variables

var generator : AsyncGenerator[Union[NoneType, Tuple[Literal[<ResponseCodeEnum.VALUE: 1>], Any]], Union[NoneType, Tuple[StatusCodeEnum, ExcInfo]]]

var generator_factory : Callable[[Any], AsyncGenerator[Union[NoneType, Tuple[Literal[<ResponseCodeEnum.VALUE: 1>], Any]], Union[NoneType, Tuple[StatusCodeEnum, ExcInfo]]]]

var ready : bool

Inherited members

• BaseDispatcher:
  • afterDispatch
  • afterExecution
  • afterTargetExec
  • beforeDispatch
  • beforeExecution
  • beforeTargetExec
  • catch
  • mixin
  • onActive

class DispatcherContextManager (generator_factory: Callable, args=None, kwargs=None)

所有非单函数型 Dispatcher 的基类, 用于为参数解析提供可扩展的支持.

Expand source code

class DispatcherContextManager(BaseDispatcher):
    generator_factory: Callable[[Any], T_DispatcherContextManager]
    generator: T_DispatcherContextManager
    ready: bool = False

    def __init__(self, generator_factory: Callable, args=None, kwargs=None) -> None:
        self.generator_factory = generator_factory
        self.args = args or ()
        self.kwargs = kwargs or {}

    def beforeExecution(self, interface: "DispatcherInterface"):
        self.generator = self.generator_factory(*self.args, **self.kwargs)
        next(self.generator)
        self.generator.send(interface)
        self.ready = True

    def catch(self, interface: "DispatcherInterface"):
        if not self.ready:
            return
        status, value = self.generator.send((StatusCodeEnum.DISPATCHING, None))
        if status is ResponseCodeEnum.VALUE:
            return value

    def afterDispatch(
        self,
        interface: "DispatcherInterface",
        exception: Optional[Exception],
        tb: Optional[TracebackType],
    ):
        self.ready = False
        try:
            if not tb:
                self.generator.send((StatusCodeEnum.DISPATCH_COMPLETED, None))
            else:
                self.generator.send(
                    (StatusCodeEnum.DISPATCH_EXCEPTION, ExcInfo(exception, tb))
                )
        except StopIteration:
            pass

    def afterExecution(
        self,
        interface: "DispatcherInterface",
        exception: Optional[Exception],
        tb: Optional[TracebackType],
    ):
        try:
            if not tb:
                self.generator.send((StatusCodeEnum.EXECUTION_COMPLETED, None))
            else:
                self.generator.send(
                    (StatusCodeEnum.EXECUTION_EXCEPTION, ExcInfo(exception, tb))
                )
        except StopIteration:
            pass

Ancestors

• BaseDispatcher

Class variables

var generator : Generator[Union[NoneType, Tuple[Literal[<ResponseCodeEnum.VALUE: 1>], Any]], Union[NoneType, Tuple[StatusCodeEnum, ExcInfo]], NoneType]

var generator_factory : Callable[[Any], Generator[Union[NoneType, Tuple[Literal[<ResponseCodeEnum.VALUE: 1>], Any]], Union[NoneType, Tuple[StatusCodeEnum, ExcInfo]], NoneType]]

var ready : bool

Inherited members

• BaseDispatcher:
  • afterDispatch
  • afterExecution
  • afterTargetExec
  • beforeDispatch
  • beforeExecution
  • beforeTargetExec
  • catch
  • mixin
  • onActive

class ExcInfo (exception: Exception, traceback: traceback)

ExcInfo(exception, traceback)

Expand source code

class ExcInfo(NamedTuple):
    exception: Exception
    traceback: TracebackType

Ancestors

• builtins.tuple

Instance variables

var exception : Exception

Alias for field number 0

var traceback : traceback

Alias for field number 1

class ResponseCodeEnum (value, names=None, *, module=None, qualname=None, type=None, start=1)

An enumeration.

Expand source code

class ResponseCodeEnum(IntEnum):
    VALUE = 1

Ancestors

• enum.IntEnum
• builtins.int
• enum.Enum

Class variables

var VALUE

class StatusCodeEnum (value, names=None, *, module=None, qualname=None, type=None, start=1)

An enumeration.

Expand source code

class StatusCodeEnum(IntEnum):
    DISPATCHING = 1

    DISPATCH_COMPLETED = 2
    EXECUTION_COMPLETED = 3

    DISPATCH_EXCEPTION = 4
    EXECUTION_EXCEPTION = 5

Ancestors

• enum.IntEnum
• builtins.int
• enum.Enum

Class variables

var DISPATCHING

var DISPATCH_COMPLETED

var DISPATCH_EXCEPTION

var EXECUTION_COMPLETED

var EXECUTION_EXCEPTION