sanic.signals.Event#

Event names for the SignalRouter

Inherits from: Enum

class Event(value, names = None, module = None, qualname = None, type = None, start = 1, boundary = None)

sanic.signals.Signal#

A Route that is used to dispatch signals to handlers

Inherits from: Route

class Signal(router, raw_path: <class 'str'>, name: <class 'str'>, handler: typing.Callable[..., typing.Any], methods: typing.Union[typing.Sequence[str], typing.FrozenSet[str]], requirements: typing.Optional[typing.Dict[str, typing.Any]] = None, strict: <class 'bool'> = False, unquote: <class 'bool'> = False, static: <class 'bool'> = False, regex: <class 'bool'> = False, overloaded: <class 'bool'> = False, priority: <class 'int'> = 0)

add_parameter#

def add_parameter(self, idx: <class 'int'>, name: <class 'str'>, raw_path: <class 'str'>, label: <class 'str'>, cast: typing.Type, pattern = None, param_info_class = <class 'sanic_routing.patterns.ParamInfo'>)

defined_params#

@property
def defined_params(self)

finalize#

def finalize(self)

parse_parameter_string#

Parse a parameter string into its constituent name, type, and

def parse_parameter_string(self, parameter_string: <class 'str'>)

pattern

For example:

parse_parameter_string('<param_one:[A-z]>')` -> ('param_one', '[A-z]', <class 'str'>, '[A-z]')
Parameters
parameter_string

String to parse

raw_path#

The raw path from the route definition

@property
def raw_path(self)

reset#

def reset(self)

segments#

Same as :py:attr:~sanic_routing.route.Route.parts except

@property
def segments(self): -> typing.Tuple[str, ...]

generalized so that any dynamic parts do not include param keys since they have no impact on routing.

uri#

Since :py:attr:~sanic_routing.route.Route.path does NOT

@property
def uri(self)

include a preceding '/', this adds it back.

sanic.signals.SignalGroup#

A RouteGroup that is used to dispatch signals to handlers

Inherits from: RouteGroup

class SignalGroup(routes): -> None

depth#

The number of parts in :py:attr:parts

@property
def depth(self): -> int

dynamic_path#

@property
def dynamic_path(self): -> bool

finalize#

def finalize(self)

merge#

The purpose of merge is to group routes with the same path, but

def merge(self, group: RouteGroup, overwrite: bool = False, append: bool = False): -> None

declarared individually. In other words to group these:

@app.get("/path/to")
def handler1(...):
    ...

@app.post("/path/to")
def handler2(...):
    ...

The other main purpose is to look for conflicts and raise RouteExists

A duplicate route is when:

  1. They have the same path and any overlapping methods; AND
  2. If they have requirements, they are the same
Parameters
group
RouteGroup

Incoming route group

overwrite
bool, optional

whether to allow an otherwise duplicate route group to overwrite the existing, if True will not raise exception on duplicates, defaults to False

append
bool, optional

whether to allow an otherwise duplicate route group to append its routes to the existing route group, defaults to False

Raises
RouteExists

Raised when there is a duplicate

methods#

@property
def methods(self): -> FrozenSet[str]

prioritize_routes#

Sorts the routes in the group by priority

def prioritize_routes(self): -> None

requirements#

@property
def requirements(self): -> List[Requirements]

reset#

def reset(self)

routes#

@property
def routes(self): -> Sequence[Route]

sanic.signals.SignalRouter#

A BaseRouter that is used to dispatch signals to handlers

Inherits from: BaseRouter, ABC

class SignalRouter(): -> None

add#

def add(self, handler: SignalHandler, event: Union[str, Enum], condition: Optional[Dict[str, Any]] = None, exclusive: bool = True, priority: int = 0): -> Signal

dispatch#

Dispatch a signal to all handlers that match the event

async def dispatch(self, event: Union[str, Enum], context: Optional[Dict[str, Any]] = None, condition: Optional[Dict[str, str]] = None, fail_not_found: bool = True, inline: bool = False, reverse: bool = False): -> Union[asyncio.Task, Any]

Parameters
event
str

The event to dispatch

context
Optional[Dict[str, Any]]

A dictionary of context to pass to the handlers. Defaults to None.

condition
Optional[Dict[str, str]]

A dictionary of conditions to match against the handlers. Defaults to None.

fail_not_found
bool

Whether to raise an exception if no handlers are found. Defaults to True.

inline
bool

Whether to run the handlers inline. An inline run means it will return the value of the signal handler. When False (which is the default) the signal handler will run in a background task. Defaults to False.

reverse
bool

Whether to run the handlers in reverse order. Defaults to False.

Return
Union[asyncio.Task, Any]

If inline is True then the return value of the signal handler will be returned. If inline is False then an asyncio.Task will be returned.

Raises
RuntimeError

If the signal is dispatched outside of an event loop

finalize#

Finalize the router and compile the routes

def finalize(self, do_compile: bool = True, do_optimize: bool = False)

Parameters
do_compile
bool

Whether to compile the routes. Defaults to True.

do_optimize
bool

Whether to optimize the routes. Defaults to False.

Return
SignalRouter

The router

Raises
RuntimeError

If the router is finalized outside of an event loop

find_route#

@property
def find_route(self)

format_event#

Ensure event strings in proper format

@staticmethod
def format_event(event: Union[str, Enum]): -> str

Parameters
event
str

event string

Return
str

formatted event string

get#

Get the handlers for a signal

def get(self, event: Union[str, Enum], condition: Optional[Dict[str, str]] = None)

Parameters
event
str

The event to get the handlers for

condition
Optional[Dict[str, str]]

A dictionary of conditions to match against the handlers. Defaults to None.

Return
Tuple[SignalGroup, List[SignalHandler], Dict[str, Any]]

A tuple of the SignalGroup that matched, a list of the handlers that matched, and a dictionary of the params that matched

Raises
NotFound

If no handlers are found

get_waiter#

def get_waiter(self, event: Union[str, Enum], condition: Optional[Dict[str, Any]] = None, exclusive: bool = True): -> Optional[SignalWaiter]

groups#

@property
def groups(self)

matchers#

@property
def matchers(self)

register_pattern#

Add a custom parameter type to the router. The cast should raise a

def register_pattern(self, label: <class 'str'>, cast: typing.Callable[[str], typing.Any], pattern: typing.Union[typing.Pattern, str], param_info_class: typing.Type[sanic_routing.patterns.ParamInfo] = <class 'sanic_routing.patterns.ParamInfo'>)

ValueError if it is an incorrect type. The order of registration is important if it is possible that a single value could pass multiple pattern types. Therefore, patterns are tried in the REVERSE order of registration. All custom patterns will be evaluated before any built-in patterns.

Parameters
label
str

The parts that is used to signify the type: example

cast
t.Callable[[str], t.Any]

The callable that casts the value to the desired type, or fails trying

pattern
Union[t.Pattern, str]

A regular expression that could also match the path segment

reset#

def reset(self)

resolve#

def resolve(self, path: <class 'str'>, method: typing.Optional[str] = None, orig: typing.Optional[str] = None, extra: typing.Optional[typing.Dict[str, str]] = None): -> typing.Tuple[sanic_routing.route.Route, typing.Callable[..., typing.Any], typing.Dict[str, typing.Any]]

routes#

@property
def routes(self)

sanic.signals.SignalWaiter#

A record representing a future waiting for a signal

class SignalWaiter(signal: Signal, event_definition: str, trigger: str = "", requirements: Optional[Dict[str, str]] = None, exclusive: bool = True, future: Optional[asyncio.Future] = None): -> None

matches#

def matches(self, event, condition)

wait#

Block until the signal is next dispatched.

async def wait(self)

Return the context of the signal dispatch, if any.