Module graia.broadcast.builtin.decorators

Expand source code 
from typing import Any, Optional
import typing
from graia.broadcast.entities.exectarget import ExecTarget
from ..entities.decorator import Decorator
from ..entities.signatures import Force
from ..interfaces.decorator import DecoratorInterface
from ..exceptions import RequirementCrashed


class Depend(Decorator):
    pre = True
    depend_callable: ExecTarget
    cache: bool = False

    def __init__(self, callable, *, cache=False):
        self.cache = cache
        self.depend_callable = ExecTarget(callable)

    def __repr__(self) -> str:
        return "<Depend target={0}>".format(self.depend_callable)

    async def target(self, interface: DecoratorInterface):
        if self.cache:
            attempt = interface.local_storage.get(self.depend_callable)
            if attempt:
                return Force(attempt)
        result = await interface.dispatcher_interface.broadcast.Executor(
            target=self.depend_callable,
            event=interface.event,
            post_exception_event=True,
        )

        if self.cache:
            interface.local_storage[self.depend_callable] = result
        return Force(result)


class OptionalParam(Decorator):
    pre = True

    def __init__(self, origin: Any):
        self.origin = origin

    async def target(self, interface: DecoratorInterface) -> Optional[Any]:
        try:
            return Force(
                await interface.dispatcher_interface.lookup_param(
                    interface.dispatcher_interface.name,
                    interface.dispatcher_interface.annotation.__args__[0]
                    if isinstance(
                        interface.dispatcher_interface.annotation, typing._GenericAlias
                    )
                    and type(None) in interface.dispatcher_interface.annotation.__args__
                    else interface.dispatcher_interface.annotation,
                    self.origin,
                )
            )
        except RequirementCrashed:
            return Force(None)

Classes

class Depend (callable, *, cache=False)
Expand source code 
class Depend(Decorator):
    pre = True
    depend_callable: ExecTarget
    cache: bool = False

    def __init__(self, callable, *, cache=False):
        self.cache = cache
        self.depend_callable = ExecTarget(callable)

    def __repr__(self) -> str:
        return "<Depend target={0}>".format(self.depend_callable)

    async def target(self, interface: DecoratorInterface):
        if self.cache:
            attempt = interface.local_storage.get(self.depend_callable)
            if attempt:
                return Force(attempt)
        result = await interface.dispatcher_interface.broadcast.Executor(
            target=self.depend_callable,
            event=interface.event,
            post_exception_event=True,
        )

        if self.cache:
            interface.local_storage[self.depend_callable] = result
        return Force(result)

Ancestors

Class variables

var cache : bool
var depend_callable : ExecTarget
var pre : bool

Methods

async def target(self, interface: DecoratorInterface) ‑> Callable[[Any], Any]
Expand source code 
async def target(self, interface: DecoratorInterface):
    if self.cache:
        attempt = interface.local_storage.get(self.depend_callable)
        if attempt:
            return Force(attempt)
    result = await interface.dispatcher_interface.broadcast.Executor(
        target=self.depend_callable,
        event=interface.event,
        post_exception_event=True,
    )

    if self.cache:
        interface.local_storage[self.depend_callable] = result
    return Force(result)
class OptionalParam (origin: Any)
Expand source code 
class OptionalParam(Decorator):
    pre = True

    def __init__(self, origin: Any):
        self.origin = origin

    async def target(self, interface: DecoratorInterface) -> Optional[Any]:
        try:
            return Force(
                await interface.dispatcher_interface.lookup_param(
                    interface.dispatcher_interface.name,
                    interface.dispatcher_interface.annotation.__args__[0]
                    if isinstance(
                        interface.dispatcher_interface.annotation, typing._GenericAlias
                    )
                    and type(None) in interface.dispatcher_interface.annotation.__args__
                    else interface.dispatcher_interface.annotation,
                    self.origin,
                )
            )
        except RequirementCrashed:
            return Force(None)

Ancestors

Class variables

var pre : bool

Methods

async def target(self, interface: DecoratorInterface) ‑> Union[Any, NoneType]
Expand source code 
async def target(self, interface: DecoratorInterface) -> Optional[Any]:
    try:
        return Force(
            await interface.dispatcher_interface.lookup_param(
                interface.dispatcher_interface.name,
                interface.dispatcher_interface.annotation.__args__[0]
                if isinstance(
                    interface.dispatcher_interface.annotation, typing._GenericAlias
                )
                and type(None) in interface.dispatcher_interface.annotation.__args__
                else interface.dispatcher_interface.annotation,
                self.origin,
            )
        )
    except RequirementCrashed:
        return Force(None)