kyoukai.asphalt

Asphalt wrappers for Kyoukai.

Classes

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.
CtxEvent(source, topic, *, …)
HTTPRequestContext(…) The context subclass passed to all requests within Kyoukai.
KyoukaiBaseComponent(app, ip: str = , …) The base class for any component used by Kyoukai.
KyoukaiComponent(app, ip: str = , …) A component for Kyoukai.
RouteInvokedEvent(source, topic, *, …) Dispatched when a route is invoked.
RouteMatchedEvent(source, topic, *, …) Dispatched when a route is matched.
RouteReturnedEvent(source, topic, *, ctx, …) Dispatched after a route has returned.
class kyoukai.asphalt.ConnectionMadeEvent(source, topic, *, protocol)[source]

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)[source]

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: kyoukai.asphalt.HTTPRequestContext)[source]

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: kyoukai.asphalt.HTTPRequestContext)[source]

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: werkzeug.wrappers.Response)[source]

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: str = '127.0.0.1', port: int = 4444, **cfg)[source]

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)[source]

Overridden in subclasses to spawn a new server.

get_server_name()[source]
Returns:The server name of this app.
get_protocol(ctx, serv_info)[source]

Gets the protocol to use for this webserver.

class kyoukai.asphalt.KyoukaiComponent(app, ip: str = '127.0.0.1', port: int = 4444, **cfg)[source]

Bases: kyoukai.asphalt.KyoukaiBaseComponent

A component for Kyoukai. This includes the built-in HTTP server.

Changed in version 2.2: Passing run_server as False will not run the inbuilt web 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()[source]
Returns:The server name of this app.
coroutine start(self, ctx)[source]

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: asphalt.core.context.Context, request: werkzeug.wrappers.Request)[source]

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.

params = None

The route parameters for this request. Usually contained by the routing URL.

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.

add_resource(value, name='default', context_attr=None, types=())

Add a resource to this context.

This will cause a resource_added event to be dispatched.

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

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

Return type:

None

add_resource_factory(factory_callback, types, name='default', context_attr=None)

Add a resource factory to this context.

This will cause a resource_added event to be dispatched.

A resource factory is a callable that generates a “contextual” resource when it is requested by either using any of the methods get_resource(), require_resource() or request_resource() or its context attribute is accessed.

When a new resource is created in this manner, it is always bound to the context through it was requested, regardless of where in the chain the factory itself was added to.

Parameters:
  • factory_callback (Callable[[Context], Any]) – a (non-coroutine) callable that takes a context instance as argument and returns a tuple of (resource object, teardown callback)
  • types (Union[type, Sequence[Type[+CT_co]]]) – one or more types to register the generated resource as on the target context
  • name (str) – name of the resource that will be created in the target context
  • context_attr (Optional[str]) – name of the context attribute the created resource will be accessible as
Raises:

asphalt.core.context.ResourceConflict – if there is an existing resource factory for the given type/name combinations or the given context variable

Return type:

None

add_teardown_callback(callback, pass_exception=False)

Add a callback to be called when this context closes.

This is intended for cleanup of resources, and the list of callbacks is processed in the reverse order in which they were added, so the last added callback will be called first.

The callback may return an awaitable. If it does, the awaitable is awaited on before calling any further callbacks.

Parameters:
  • callback (Callable) – a callable that is called with either no arguments or with the exception that ended this context, based on the value of pass_exception
  • pass_exception (bool) – True to pass the callback the exception that ended this context (or None if the context ended cleanly)
Return type:

None

call_async(func, *args, **kwargs)

Call the given callable in the event loop thread.

This method lets you call asynchronous code from a worker thread. Do not use it from within the event loop thread.

If the callable returns an awaitable, it is resolved before returning to the caller.

Parameters:
  • func (Callable) – a regular function or a coroutine function
  • args – positional arguments to call the callable with
  • kwargs – keyword arguments to call the callable with
Returns:

the return value of the call

call_in_executor(func, *args, executor=None, **kwargs)

Call the given callable in an executor.

Parameters:
  • func (Callable) – the callable to call
  • args – positional arguments to call the callable with
  • executor (Union[Executor, str, None]) – either an Executor instance, the resource name of one or None to use the event loop’s default executor
  • kwargs – keyword arguments to call the callable with
Return type:

Awaitable[+T_co]

Returns:

an awaitable that resolves to the return value of the call

coroutine close(self, exception=None)

Close this context and call any necessary resource teardown callbacks.

If a teardown callback returns an awaitable, the return value is awaited on before calling any further teardown callbacks.

All callbacks will be processed, even if some of them raise exceptions. If at least one callback raised an error, this method will raise a TeardownError at the end.

After this method has been called, resources can no longer be requested or published on this context.

Parameters:exception (Optional[BaseException]) – the exception, if any, that caused this context to be closed
Raises:TeardownError – if one or more teardown callbacks raise an exception
Return type:None
closed

Return True if the context has been closed, False otherwise.

Return type:bool
context_chain

Return a list of contexts starting from this one, its parent and so on.

Return type:List[Context]
get_resource(type, name='default')

Look up a resource in the chain of contexts.

Parameters:
  • type (type) – type of the requested resource
  • name (str) – name of the requested resource
Returns:

the requested resource, or None if none was available

loop

Return the event loop associated with this context.

Return type:AbstractEventLoop
parent

Return the parent context, or None if there is no parent.

Return type:Optional[Context]
coroutine request_resource(self, type, name='default')

Look up a resource in the chain of contexts.

This is like get_resource() except that if the resource is not already available, it will wait for one to become available.

Parameters:
  • type (type) – type of the requested resource
  • name (str) – name of the requested resource
Returns:

the requested resource

require_resource(type, name='default')

Look up a resource in the chain of contexts and raise an exception if it is not found.

This is like get_resource() except that instead of returning None when a resource is not found, it will raise ResourceNotFound.

Parameters:
  • type (type) – type of the requested resource
  • name (str) – name of the requested resource
Returns:

the requested resource

Raises:

asphalt.core.context.ResourceNotFound – if a resource of the given type and name was not found

threadpool(executor=None)

Return an asynchronous context manager that runs the block in a (thread pool) executor.

Parameters:executor (Union[Executor, str, None]) – either an Executor instance, the resource name of one or None to use the event loop’s default executor
Returns:an asynchronous context manager
url_for(endpoint, *, method=None, **kwargs)[source]

A context-local version of url_for.

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