kyoukai.asphalt

Asphalt wrappers for Kyoukai.

Functions

resolve_reference(ref)

Return the object pointed to by ref.

Classes

Blueprint(name[, parent, prefix, ...])	A Blueprint class contains a Map of URL rules, which is checked and ran for every
Component	This is the base class for all Asphalt components.
ConnectionLostEvent(source, topic, *, protocol)	Dispatched when a connection is lost from the server.
ConnectionMadeEvent(source, topic, *, protocol)	Dispatched when a connection is made to the server.
Context([parent, default_timeout])	Contexts give request handlers and callbacks access to resources.
CtxEvent(source, topic, *, ctx)
Event(source, topic)	The base class for all events.
HTTPRequestContext(parent, request)	The context subclass passed to all requests within Kyoukai.
KyoukaiBaseComponent(app[, ip, port])	The base class for any component used by Kyoukai.
KyoukaiComponent(app[, ip, port])	A component for Kyoukai.
Request(environ[, populate_request, shallow])	Full featured request object implementing the following mixins:
Response([response, status, headers, ...])	Full featured response object implementing the following mixins:
Route(function[, reverse_hooks, ...])	A route object is a wrapped function.
RouteInvokedEvent(source, topic, *, ctx)	Dispatched when a route is invoked.
RouteMatchedEvent(source, topic, *, ctx)	Dispatched when a route is matched.
RouteReturnedEvent(source, topic, *, ctx, result)	Dispatched after a route has returned.
Rule(string[, defaults, subdomain, methods, ...])	A Rule represents one URL pattern.
Signal(event_class, *[, source, topic])	Declaration of a signal that can be used to dispatch events.
partial	partial(func, *args, **keywords) - new function with partial application

class kyoukai.asphalt.ConnectionMadeEvent(source, topic, *, protocol)

Bases: asphalt.core.event.Event

Dispatched when a connection is made to the server.

This does NOT fire when using WSGI workers.

This has the protocol as the protocol attribute.

utc_timestamp

Return a timezone aware datetime corresponding to the time variable, using the UTC timezone.

Return type:datetime

class kyoukai.asphalt.ConnectionLostEvent(source, topic, *, protocol)

Bases: kyoukai.asphalt.ConnectionMadeEvent

Dispatched when a connection is lost from the server.

This does NOT fire when using WSGI workers.

This has the protocol as the protocol attribute.

utc_timestamp

Return a timezone aware datetime corresponding to the time variable, using the UTC timezone.

Return type:datetime

class kyoukai.asphalt.RouteMatchedEvent(source, topic, *, ctx)

Bases: kyoukai.asphalt.CtxEvent

Dispatched when a route is matched.

This has the context as the ctx attribute, and the route can be accessed via ctx.route.

utc_timestamp

Return a timezone aware datetime corresponding to the time variable, using the UTC timezone.

Return type:datetime

class kyoukai.asphalt.RouteInvokedEvent(source, topic, *, ctx)

Bases: kyoukai.asphalt.CtxEvent

Dispatched when a route is invoked.

This has the context as the ctx attribute.

utc_timestamp

Return a timezone aware datetime corresponding to the time variable, using the UTC timezone.

Return type:datetime

class kyoukai.asphalt.RouteReturnedEvent(source, topic, *, ctx, result)

Bases: kyoukai.asphalt.CtxEvent

Dispatched after a route has returned.

This has the context as the ctx attribute and the response as the result attribute.

utc_timestamp

Return a timezone aware datetime corresponding to the time variable, using the UTC timezone.

Return type:datetime

class kyoukai.asphalt.KyoukaiBaseComponent(app, ip='127.0.0.1', port=4444, **cfg)

Bases: asphalt.core.component.Component

The base class for any component used by Kyoukai.

This one does not create a Server instance; it should be used when you are using a different HTTP server backend.

app = None

The application object for a this component.

ip = None

The IP address to boot the server on.

port = None

The port to boot the server on.

cfg = None

The config file to use.

server = None

The asyncio.Server instance that is serving us today.

base_context = None

The base context for this server.

backend = None

The backend to use for the HTTP server.

coroutine start(self, ctx)

Overridden in subclasses to spawn a new server.

get_server_name()

Returns:The server name of this app.

get_protocol(ctx, serv_info)

Gets the protocol to use for this webserver.

class kyoukai.asphalt.KyoukaiComponent(app, ip='127.0.0.1', port=4444, **cfg)

Bases: kyoukai.asphalt.KyoukaiBaseComponent

A component for Kyoukai.

This includes the built-in HTTP server.

Creates a new component.

Parameters:

• app – The application object to use. This can either be the real application object, or a string that resolves to a reference for the real application object.
• ip (str) – If using the built-in HTTP server, the IP to bind to.
• port (int) – If using the built-in HTTP server, the port to bind to.
• cfg – Additional configuration.

get_server_name()

Returns:The server name of this app.

coroutine start(self, ctx)

Starts the webserver if required.

Parameters:ctx (Context) – The base context.

get_protocol(ctx, serv_info)

Gets the protocol to use for this webserver.

class kyoukai.asphalt.HTTPRequestContext(parent, request)

Bases: asphalt.core.context.Context

The context subclass passed to all requests within Kyoukai.

app = None

The Kyoukai object this request is handling.

request = None

The werkzeug.wrappers.Request object this request is handling.

route = None

The Route object this request is for.

bp = None

The Blueprint object this request is for.

rule = None

The werkzeug.routing.Rule object associated with this request.

environ = None

The WSGI environment for this request.

proto = None

The asyncio.Protocol protocol handling this connection.

get_resources(type=None, *, include_parents=True)

Return the currently published resources specific to one type or all types.

Parameters:

• type (Union[str, type, None]) – type of the resources to return, or None to return all resources
• include_parents (bool) – include the resources from parent contexts

Return type:

Sequence[Resource]

parent

Return the parent of this context or None if there is no parent context.

Return type:Optional[Context]

publish_lazy_resource(creator, types, alias='default', context_attr=None)

Publish a “lazy” or “contextual” resource and dispatch a resource_published event.

Instead of a concrete resource value, you supply a creator callable which is called with a context object as its argument when the resource is being requested either via request_resource() or by context attribute access. The return value of the creator callable will be cached so the creator will only be called once per context instance.

If the creator callable is a coroutine function or returns an awaitable, it is resolved before storing the resource value and returning it to the requester. Note that this will NOT work when a context attribute has been specified for the resource.

Parameters:

• creator (Callable[[Context], Any]) – a callable taking a context instance as argument
• types (Union[type, Iterable[Union[str, type]]]) – type(s) to register the resource as
• alias (str) – name of this resource (unique among all its registered types)
• context_attr (Optional[str]) – name of the context attribute this resource will be accessible as

Return type:

Resource

Returns:

the resource handle

Raises:

asphalt.core.context.ResourceConflict – if there is an existing resource creator for the given types or context variable

publish_resource(value, alias='default', context_attr=None, *, types=())

Publish a resource and dispatch a resource_published event.

Parameters:

• value – the actual resource value
• alias (str) – name of this resource (unique among all its registered types)
• context_attr (Optional[str]) – name of the context attribute this resource will be accessible as
• types (Union[type, Iterable[Union[str, type]]]) – type(s) to register the resource as (omit to use the type of value)

Return type:

Resource

Returns:

the resource handle

Raises:

asphalt.core.context.ResourceConflict – if the resource conflicts with an existing one in any way

remove_resource(resource)

Remove the given resource from the collection and dispatch a resource_removed event.

Parameters:resource (Resource) – the resource to be removed Raises:LookupError – the given resource was not in the collection

coroutine request_resource(self, type, alias='default', *, timeout=None)

Request a resource matching the given type and alias.

If no such resource was found, this method will wait timeout seconds for it to become available. The timeout does not apply to resolving awaitables created by lazy resource creators.

Parameters:

• type (Union[str, type]) – type of the requested resource
• alias (str) – alias of the requested resource
• timeout (Union[int, float, None]) – the timeout (in seconds; omit to use the context’s default timeout)

Returns:

the value contained by the requested resource (NOT a Resource instance)

Raises:

asphalt.core.context.ResourceNotFound – if the requested resource does not become available in the allotted time

url_for(endpoint, *, method, **kwargs)

A context-local version of url_for.

For more information, see the documentation on url_for().