Kyoukai Autodoc

This is automatically generated API documentation for the kyoukai module.

Kyoukai is an async web framework for Python 3.5 and above.

app The core application.
backends Various backends that interface with the Kyoukai application.
asphalt Asphalt wrappers for Kyoukai.
blueprint A blueprint is a container - a collection of routes.
route Routes are wrapped function objects that are called upon a HTTP request.
routegroup Route groups are classes that allow you to group a set of routes together.
testing Testing helpers for Kyoukai.
util Misc utilities for usage inside the framework.
class kyoukai.Kyoukai(application_name, *, server_name=None, **kwargs)[source]

Bases: object

The Kyoukai type is the core of the Kyoukai framework, and the core of your web application based upon the Kyoukai framework. It acts as a central router and request processor that takes in requests from the protocols and returns responses.

The application name is currently unused, but it is good practice to set it correctly anyway in case it is used in future editions of Kyoukai.

You normally create an application instance inside your component file, like so:

from kyoukai.app import Kyoukai

... # setup code

kyk = Kyoukai("my_app")
kyk.register_blueprint(whatever)

... # other setup

class MyContainer(ContainerComponent):
    async def start(self, ctx):
        self.add_component('kyoukai', KyoukaiComponent, ip="127.0.0.1", port=4444,
                           app="app:app")

Of course, you can also embed Kyoukai inside another app, by awaiting Kyoukai.start().

Parameters:
  • application_name (str) – The name of the application that is being created. This is passed to the Blueprint being created as the root blueprint.
  • server_name (Optional[str]) – Keyword-only. The SERVER_NAME to use inside the fake WSGI environment created for url_for, if applicable.
  • host_matching – Should host matching be enabled? This will be implicitly True if host is not None.
  • host – The host used for host matching, to be passed to the root Blueprint. By default, no host is used, so all hosts are matched on the root Blueprint.
  • application_root – Keyword-only. The APPLICATION_ROOT to use inside the fake WSGI environment created for url_for, if applicable.
  • loop – Keyword-only. The asyncio event loop to use for this app. If no loop is specified it, will be automatically fetched using asyncio.get_event_loop().
  • request_class – Keyword-only. The custom request class to instantiate requests with.
  • response_class – Keyword-only. The custom response class to instantiate responses with.
finalize()[source]

Finalizes the app and blueprints.

This will calculate the current werkzeug.routing.Map which is required for routing to work.

coroutine handle_httpexception(self, ctx, exception, environ=None)[source]

Handle a HTTP Exception.

Parameters:
Return type:

Response

Returns:

A werkzeug.wrappers.Response that handles this response.

log_route(request, code)[source]

Logs a route invocation.

Parameters:
  • request (Request) – The request produced.
  • code (int) – The response code of the route.
coroutine process_request(self, request, parent_context)[source]

Processes a Request and returns a Response object.

This is the main processing method of Kyoukai, and is meant to be used by one of the HTTP server backends, and not by client code.

Parameters:
  • request (Request) – The werkzeug.wrappers.Request object to process. A new HTTPRequestContext will be provided to wrap this request inside of to client code.
  • parent_context (Context) – The asphalt.core.Context that is the parent context for this particular app. It will be used as the parent for the HTTPRequestContext.
Return type:

Response

Returns:

A werkzeug.wrappers.Response object that can be written to the client as a response.

register_blueprint(child)[source]

Registers a child blueprint to this app’s root Blueprint.

This will set up the Blueprint tree, as well as setting up the routing table when finalized.

Parameters:child (Blueprint) – The child Blueprint to add. This must be an instance of Blueprint.
request_class

alias of Request

response_class

alias of Response

root
Return type:Blueprint
Returns:The root Blueprint for the routing tree.
run(ip='127.0.0.1', port=4444, *, component=None)[source]

Runs the Kyoukai server from within your code.

This is not normally invoked - instead Asphalt should invoke the Kyoukai component. However, this is here for convenience.

coroutine start(self, ip='127.0.0.1', port=4444, *, component=None, base_context=None)[source]

Runs the Kyoukai component asynchronously.

This will bypass Asphalt’s default runner, and allow you to run your app easily inside something else, for example.

Parameters:
  • ip (str) – The IP of the built-in server.
  • port (int) – The port of the built-in server.
  • component – The component to start the app with. This should be an instance of kyoukai.asphalt.KyoukaiComponent.
  • base_context (Optional[Context]) – The base context that the HTTPRequestContext should be started with.
class kyoukai.HTTPRequestContext(parent, request)[source]

Bases: asphalt.core.context.Context

The context subclass passed to all requests within Kyoukai.

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

A context-local version of url_for.

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

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

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_protocol(ctx, serv_info)

Gets the protocol to use for this webserver.

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.
class kyoukai.Blueprint(name, parent=None, prefix='', *, host_matching=False, host=None)[source]

Bases: object

A Blueprint class contains a Map of URL rules, which is checked and ran for every

Parameters:
  • name (str) – The name of this Blueprint. This is used when generating endpoints in the finalize stage.
  • parent (Optional[Blueprint]) – The parent of this Blueprint. Parent blueprints will gather the routes of their children, and return a giant werkzeug.routing.Map object that contains all of the route maps in the children
  • prefix (str) – The prefix to be added to the start of every route name. This is inherited from parents - the parent prefix will also be added to the start of every route.
  • host_matching (bool) – Should host matching be enabled? This is implicitly True if host is non-None.
  • host (Optional[str]) – The host of the Blueprint. Used for custom subdomain routing. If this is None, then this Blueprint will be used for all hosts.
add_child(blueprint)[source]

Adds a Blueprint as a child of this one. This is automatically called when using another Blueprint as a parent.

Parameters:blueprint (Blueprint) – The blueprint to add as a child.
Return type:Blueprint
add_errorhandler(cbl, errorcode)[source]

Adds an error handler to the table of error handlers.

A blueprint can only have one error handler per code. If it doesn’t have an error handler for that code, it will try to fetch recursively the parent’s error handler.

Parameters:
  • cbl – The callable error handler.
  • errorcode (int) – The error code to handle, for example 404.
add_hook(type_, hook)[source]

Adds a hook to the current Blueprint.

Parameters:
  • type (str) – The type of hook to add (currently “pre” or “post”).
  • hook – The callable function to add as a hook.
add_route(route, routing_url, methods=('GET', ))[source]

Adds a route to the routing table and map.

Parameters:
  • route (Route) –

    The route object to add.

    This can be gotten from Blueprint.wrap_route, or by directly creating a Route object.

  • routing_url (str) –

    The Werkzeug-compatible routing URL to add this route under.

    For more information, see http://werkzeug.pocoo.org/docs/0.11/routing/.

  • methods (Iterable[str]) – An iterable of valid method this route can be called with.
Returns:

The unmodified Route object.

add_route_group(group)[source]

Adds a route group to the current Blueprint.

Parameters:group (RouteGroup) – The RouteGroup to add.
after_request(func)[source]

Convenience decorator to add a post-request hook.

before_request(func)[source]

Convenience decorator to add a pre-request hook.

errorhandler(code)[source]

Helper decorator for adding an error handler.

This is equivalent to:

route = bp.add_errorhandler(cbl, code)
Parameters:code (int) – The error handler code to use.
finalize(**map_options)[source]

Called on the root Blueprint when all Blueprints have been registered and the app is starting.

This will automatically build a werkzeug.routing.Map of werkzeug.routing.Rule objects for each Blueprint.

Note

Calling this on sub-blueprints will have no effect, apart from generating a Map. It is recommended to only call this on the root Blueprint.

Parameters:map_options – The options to pass to the created Map.
Return type:Map
Returns:The werkzeug.routing.Map created from the routing tree.
get_errorhandler(exc)[source]

Recursively acquires the error handler for the specified error.

Parameters:exc (Union[HTTPException, int]) – The exception to get the error handler for. This can either be a HTTPException object, or an integer.
Return type:Union[None, Route]
Returns:The Route object that corresponds to the error handler, or None if no error handler could be found.
get_hooks(type_)[source]

Gets a list of hooks that match the current type.

These are ordered from parent to child.

Parameters:type (str) – The type of hooks to get (currently “pre” or “post”).
Returns:An iterable of hooks to run.
get_route(endpoint)[source]

Gets the route associated with an endpoint.

Return type:Optional[Route]
host
Return type:str
Returns:The host for this Blueprint, or the host of any parent Blueprint.
match(environment)[source]

Matches with the WSGI environment.

Parameters:environment (dict) –

The environment dict to perform matching with.

You can use the environ argument of a Request to get the environment back.

Return type:Tuple[Route, Container[Any]]
Returns:A Route object, which can be invoked to return the right response, and the parameters to invoke it with.
parent
Return type:Blueprint
Returns:The parent Blueprint of this blueprint.
prefix
Return type:str
Returns:The combined prefix of this Blueprint.
route(routing_url, methods=('GET', ), **kwargs)[source]

Convenience decorator for adding a route.

This is equivalent to:

route = bp.wrap_route(func, **kwargs)
bp.add_route(route, routing_url, methods)
traverse_tree()[source]

Traverses the tree for children Blueprints.

Return type:Generator[Blueprint, None, None]
tree_routes
Return type:Generator[Route, None, None]
Returns:A generator that yields all routes from the tree, from parent to children.
url_for(environment, endpoint, *, method=None, **kwargs)[source]

Gets the URL for a specified endpoint using the arguments of the route.

This works very similarly to Flask’s url_for.

It is not recommended to invoke this method directly - instead, url_for is set on the context object that is provided to your user function. This will allow you to invoke it with the correct environment already set.

Parameters:
  • environment (dict) – The WSGI environment to use to bind to the adapter.
  • endpoint (str) – The endpoint to try and retrieve.
  • method (Optional[str]) – If set, the method to explicitly provide (for similar endpoints with different allowed routes).
  • kwargs – Keyword arguments to provide to the route.
Return type:

str

Returns:

The built URL for this endpoint.

wrap_route(cbl, *args, **kwargs)[source]

Wraps a callable in a Route.

This is required for routes to be added. :param cbl: The callable to wrap. :rtype: Route :return: A new Route object.

class kyoukai.Route(function, reverse_hooks=False, should_invoke_hooks=True, do_argument_checking=True)[source]

Bases: object

A route object is a wrapped function. They invoke this function when invoked on routing and calling.

Parameters:
  • function – The underlying callable. This can be a function, or any other callable.
  • reverse_hooks (bool) – If the request hooks should be reversed for this request (i.e child to parent.)
  • should_invoke_hooks (bool) – If request hooks should be invoked. This is automatically False for error handlers.
  • do_argument_checking (bool) – If argument type and name checking is enabled for this route.
add_hook(type_, hook)[source]

Adds a hook to the current Route.

Parameters:
  • type (str) – The type of hook to add (currently “pre” or “post”).
  • hook – The callable function to add as a hook.
after_request(func)[source]

Convenience decorator to add a pre-request hook.

before_request(func)[source]

Convenience decorator to add a post-request hook.

check_route_args(params=None)[source]

Checks the arguments for a route.

Parameters:params (Optional[dict]) – The parameters passed in, as a dict.
Raises:TypeError – If the arguments passed in were not correct.
create_rule()[source]

Creates the rule object used by this route.

Return type:Rule
Returns:A new werkzeug.routing.Rule that is to be used for this route.
get_endpoint_name(bp=None)[source]

Gets the endpoint name for this route.

get_hooks(type_)[source]

Gets the hooks for the current Route for the type.

Parameters:type (str) – The type to get.
Returns:A list of callables.
coroutine invoke(self, ctx, params=None)[source]

Invokes a route. This will run the underlying function.

Parameters:
Return type:

Response

Returns:

The result of the route’s function.

coroutine invoke_function(self, ctx, pre_hooks, post_hooks, params)[source]

Invokes the underlying callable.

This is for use in chaining routes. :param ctx: The HTTPRequestContext to use for this route. :type pre_hooks: list :param pre_hooks: A list of hooks to call before the route is invoked. :type post_hooks: list :param post_hooks: A list of hooks to call after the route is invoked. :param params: The parameters to pass to the function. :return: The result of the invoked function.

class kyoukai.RouteGroup[source]

Bases: object

A route group is a class that contains multiple methods that are decorated with the route decorator. They produce a blueprint that can be added to the tree that includes all methods in the route group.

class MyGroup(RouteGroup, url_prefix="/api/v1"):
    def __init__(self, something: str):
        self.something = something
        
    @route("/ping")
    async def ping(self, ctx: HTTPRequestContext):
        return '{"response": self.something}'

Blueprint parameters can be passed in the class call.

To add the route group as a blueprint, use Blueprint.add_route_group(MyGroup, *args, **kwargs)().

class kyoukai.TestKyoukai(*args, base_context=None, **kwargs)[source]

Bases: kyoukai.app.Kyoukai

A special subclass that allows you to easily test your Kyoukai-based app.

Parameters:base_context (Optional[Context]) – The base context to use for all request testing.
finalize()

Finalizes the app and blueprints.

This will calculate the current werkzeug.routing.Map which is required for routing to work.

coroutine handle_httpexception(self, ctx, exception, environ=None)

Handle a HTTP Exception.

Parameters:
Return type:

Response

Returns:

A werkzeug.wrappers.Response that handles this response.

coroutine inject_request(self, headers, url, method='GET', body=None)[source]

Injects a request into the test client.

This will automatically create the correct context.

Parameters:
  • headers (dict) – The headers to use.
  • body (Optional[str]) – The body to use.
  • url (str) – The URL to use.
  • method (str) – The method to use.
Return type:

Response

Returns:

The result.

log_route(request, code)

Logs a route invocation.

Parameters:
  • request (Request) – The request produced.
  • code (int) – The response code of the route.
coroutine process_request(self, request, parent_context)

Processes a Request and returns a Response object.

This is the main processing method of Kyoukai, and is meant to be used by one of the HTTP server backends, and not by client code.

Parameters:
  • request (Request) – The werkzeug.wrappers.Request object to process. A new HTTPRequestContext will be provided to wrap this request inside of to client code.
  • parent_context (Context) – The asphalt.core.Context that is the parent context for this particular app. It will be used as the parent for the HTTPRequestContext.
Return type:

Response

Returns:

A werkzeug.wrappers.Response object that can be written to the client as a response.

register_blueprint(child)

Registers a child blueprint to this app’s root Blueprint.

This will set up the Blueprint tree, as well as setting up the routing table when finalized.

Parameters:child (Blueprint) – The child Blueprint to add. This must be an instance of Blueprint.
request_class

alias of Request

response_class

alias of Response

root
Return type:Blueprint
Returns:The root Blueprint for the routing tree.
run(ip='127.0.0.1', port=4444, *, component=None)

Runs the Kyoukai server from within your code.

This is not normally invoked - instead Asphalt should invoke the Kyoukai component. However, this is here for convenience.

coroutine start(self, ip='127.0.0.1', port=4444, *, component=None, base_context=None)

Runs the Kyoukai component asynchronously.

This will bypass Asphalt’s default runner, and allow you to run your app easily inside something else, for example.

Parameters:
  • ip (str) – The IP of the built-in server.
  • port (int) – The port of the built-in server.
  • component – The component to start the app with. This should be an instance of kyoukai.asphalt.KyoukaiComponent.
  • base_context (Optional[Context]) – The base context that the HTTPRequestContext should be started with.
testing_bp()[source]

Context handler that allows with TestKyoukai.testing_bp() as bp:

You can then register items onto this new root blueprint until __exit__, which will then destroy the blueprint.

Return type:_TestingBpCtxManager
classmethod wrap_existing_app(other_app, base_context=None)[source]

Wraps an existing app in a test frame.

This allows easy usage of writing unit tests:

# main.py
kyk = Kyoukai("my_app")

# test.py
testing = TestKyoukai.wrap_existing_app(other_app)
# use testing as you would normally
Parameters:
  • other_app (Kyoukai) –

    The application object to wrap. Internally, this creates a new instance of ourselves, then sets the process_request of the subclass to the copied object.

    This means whenever inject_request is called, it will use the old app’s process_request to run with, which will use the environment of the previous instance.

    Of course, if the old app has any side effects upon process_request, these side effects will happen when the testing application runs as well, as the old app is completely copied over.

  • base_context (Optional[Context]) – The base context to use for this.