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: str, *, server_name: str = 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.

    This is used in url_for, for example, to get the endpoint of routes registered to 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.
  • context_class – Keyword-only. The Context subclass to use when creating a context. Defaults to HTTPRequestContext.
finalize(**map_options)[source]

Finalizes the app and blueprints.

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

Parameters:map_options – The options to pass to the Map for routing.
Return type:Map
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 KyoukaiComponent.
  • base_context (Optional[Context]) – The base context that the HTTPRequestContext should be started with.
class kyoukai.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.

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().

class kyoukai.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_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: str, parent: typing.Union[kyoukai.blueprint.Blueprint, NoneType] = None, prefix: str = '', *, host_matching: bool = False, host: str = None)[source]

Bases: object

A Blueprint is a “route container” - it contains 0 to N routes, and 0 to N child Blueprints that inherit from the parent.

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, startcode, endcode=None, step=None)[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.
  • startcode (int) – The error code to handle, for example 404. This also represents the start of an error range, if endcode is not None.
  • endcode (Optional[int]) – The end of the error code range to handle. Error handlers will be added for all requests between startcode and endcode.
  • step (Optional[int]) – The step for the error handler range.
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', 'HEAD'))[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 (Sequence[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.

computed_prefix
Return type:str
Returns:The combined prefix (parent + ours) of this Blueprint.

New in version 2.2.0.

errorhandler(code, endcode=None, step=None)[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.
  • endcode (Optional[int]) – The end of the error code range to handle. Error handlers will be added for all requests between code and endcode. If this is not provided, only one code will be handled.
  • step (Optional[int]) – The step for the error handler range.
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.

Changed in version 2.2.0: This now uses submounts instead of a giant rule amalgamation.

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

Gets the werkzeug.routing.Submount for this Blueprint.

New in version 2.2.0.

Return type:Submount
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.

Warning

You should not be using this method yourself.

Changed in version 2.2.0: This will now return the werkeug.routing.Rule as well.

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], Rule]
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 prefix of this Blueprint.

Changed in version 2.2.0: Moved prefix combination to computed_prefix.

route(routing_url, methods=('GET', 'HEAD'), **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)

Changed in version 2.2.0: Now accepts a Route as the function to decorate - this will add a new routing url and method pair to Route.add_route().

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.

Parameters:cbl – The callable to wrap.
Return type:Route
Returns:A new Route object.
class kyoukai.Route(function, *, reverse_hooks: bool = False, should_invoke_hooks: bool = True, do_argument_checking: bool = True, endpoint: str = None)[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.
  • endpoint (Optional[str]) – The custom endpoint 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.
add_path(url, methods=('GET', 'HEAD'))[source]

Adds a path to the current set of paths for this route.

Parameters:
  • url (str) – The routing URL to add.
  • methods (Sequence[str]) – An iterable of methods to use for this path.

The URL and methods will be added as a pair.

after_request(func)[source]

Convenience decorator to add a post-request hook.

before_request(func)[source]

Convenience decorator to add a pre-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.
get_endpoint_name(bp=None)[source]

Gets the endpoint name for this route.

Parameters:bp – The Blueprint to use for name calculation.
Return type:str
Returns:The endpoint that can be used.
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.
get_submount()[source]
Return type:Submount
Returns:A submount that represents this route.

New in version 2.2.0.

Changed in version 2.x.x: Changed from getting a list of rules to a single submount object.

coroutine invoke(self, ctx, args=(), 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.

Parameters:
  • ctx – The HTTPRequestContext to use for this route.
  • pre_hooks (list) – A list of hooks to call before the route is invoked.
  • post_hooks (list) – A list of hooks to call after the route is invoked.
  • params – The parameters to pass to the function.
Returns:

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, 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: asphalt.core.context.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(**map_options)

Finalizes the app and blueprints.

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

Parameters:map_options – The options to pass to the Map for routing.
Return type:Map
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 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(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.