incant 22.1.0

Simple Dependencies

After a while, your colleague says they require the IP address of the incoming request. You explain the Quart API for this (request.remote_addr), but your colleague is adamant about following best practices (avoiding global variables) - they want it as an argument to their handler. They also want it as an instance of Python’s ipaddress.IPv4Address. Their handler looks like this:

@app.get("/ip")
async def ip_address_handler(source_ip: IPv4Address) -> str:
    return f"Your address is {source_ip}"

Well, looks like you’ve got your work cut out for you.

At the top of the file, you import and prepare an incant.Incanter instance.

from incant import Incanter

incanter = Incanter()

You decide to write a function to get the address from the request, and to register it with your Incanter to be matched by type.

from ipaddress import IPv4Address
from quart import request

@incanter.register_by_type
def get_ip_address() -> IPv4Address:
    # In Quart (like in Flask), the request is accessed through a global proxy
    return IPv4Address(request.remote_addr)

This means any function invoked through the Incanter will have any parameters annotated as IPv4Address satisfied by calling the get_ip_address dependency factory.

You contemplate how to get this information to the ip_address_handler, and choose to write a simple decorator (yay Python!). Your colleague agrees, but (citing consistency) wants the decorator to be applied to all handlers going forward.

(You could solve this more elegantly by subclassing the quart.Quart class but forgo this as this is an incant tutorial, not a Quart one.)

You rub your hands and mutter “Let’s roll” to yourself.

from functools import wraps

def quickapi(handler):
    @wraps(handler)
    async def wrapper():
        return await incanter.ainvoke(handler)

    return wrapper

incanter.ainvoke (the async version of invoke) does what you want - invokes the coroutine you give it while satisfying its arguments from its internal dependency factories.

Then you just apply the decorators to both existing handlers.

@app.get("/ip")
@quickapi
async def ip_address_handler(source_ip: IPv4Address) -> str:
    return f"Your address is {source_ip}"

Passing in Dependencies from the Outside

Some time later, another colleague approaches you asking for path variables to be provided to their handler. Their handler needs to look like this:

@app.get("/even-or-odd/<int:integer>")
@quickapi
async def even_or_odd_handler(integer: int) -> str:
    return "odd" if integer % 2 != 0 else "even"

Quart provides path parameters like this to handlers as kwargs, so you modify the quickapi decorator a little:

def quickapi(handler):
    @wraps(handler)
    async def wrapper(**kwargs):
        return await incanter.ainvoke(handler, **kwargs)

    return wrapper

The decorator simply receives them and passes them along to the handler. This works because incant will use arguments provided to invoke if it cannot satisfy a parameter using its internal dependency factories.

Another day of earning your keep!

The Magic of incant

Some time later, another colleague approaches you asking for a logger to be provided to their handler. They want to use structured logging, and they want the logger to already be bound with the name of the handler. You think the proposal is well-thought-out and want to use the logger yourself to log every request.

Here’s what they want their handler to look like:

@app.get("/log")
@quickapi
async def logging_handler(log: BoundLogger) -> str:
    log.info("Hello from the log handler")
    return "Response after logging"

You reach for the trusty structlog library and set it up.

from structlog.stdlib import BoundLogger, get_logger

logger = get_logger()  # Useful to have a module-scoped one first.

You change the quickapi decorator to create and use a logger with the current handler name:

def quickapi(handler):
    log = logger.bind(handler=handler.__name__)

    @wraps(handler)
    async def wrapper(**kwargs):
        log.info("Processing")
        return await incanter.ainvoke(handler, **kwargs)

    return wrapper

You can’t make the logger a dependency within the Incanter though, since it depends on handler-specific data. (You could have a separate incanter for each handler, but that’s inefficient.)

If the incanter cannot find a dependency to fulfil a parameter, you need to provide it yourself - just like with the path parameters. Since the index and ip_address_handler don’t require the logger, we can keep invoking them as before. However, the logging_handler handler requires it. Without changes, invoking the handler will let you know:

TypeError: invoke_logging_handler() missing 1 required positional argument: 'log'

You change the quickapi decorator to use Incanter.aincant (the async version of Incanter.incant) and always pass in the logger instance. incant is meant for cases like this, forwarding the parameters if they are needed and skipping them otherwise. Since incant doesn’t itself call invoke, you prepare it yourself beforehand.

def quickapi(handler):
    log = logger.bind(handler=handler.__name__)

    prepared = incanter.prepare(handler)

    @wraps(handler)
    async def wrapper(**kwargs):
        log.info("Processing")
        return await incanter.aincant(prepared, log=log, **kwargs)

    return wrapper

Since you’re passing in the logger using kwargs, it will match (after trying name+type) any parameter named log.

Nested Dependencies

A colleague is working on an authentication system for your product. They have a function that takes a cookie (named session_token) and produces an instance of your user model.

from attrs import define

@define
class User:
    """The user model."""
    username: str

async def current_user(session_token: str) -> User:
    # Complex black magic goes here, immune to timing attacks.
    return User("admin")

They want to be able to use this user model in their handler.

@app.get("/user")
@quickapi
async def user_handler(user: User, log) -> str:
    log.info("Chilling here", user=repr(user))
    return "After the user handler"

You can use their current_user coroutine directly as a dependency factory:

incanter.register_by_type(current_user)

but this still leaves the issue of getting the cookie from somewhere. You define a dependency factory for the session token cookie:

# We're using a lambda, so we pass in the `name` explicitly.
incanter.register_by_name(lambda: request.cookies['session_token'], name="session_token")

Because of how request.cookies works on Quart, this handler will respond with 400 if the cookie is not present, or run the handler otherwise. But only for the handlers that require the User dependency.

Pretty cool!

Async Context Managers

A colleague of yours has heard of this newfangled concept of structured concurrency, and insists on trying it out. You offer to let them use TaskGroups from the quattro library.

Their handler looks like this:

from quattro import TaskGroup

@app.get("/taskgroup")
@quickapi
async def taskgroup_handler(tg: TaskGroup, log: BoundLogger) -> str:
    async def inner():
        log.info("Using structured concurrency, not leaking tasks")

    tg.create_task(inner())
    return "nice"

You don’t feel particularly challenged, as incant support async context managers out of the box and the only thing you need to do is:

incanter.register_by_type(TaskGroup, is_context_manager="async")

Forced Dependencies

Yesterday you’ve had an outage due to being featured by an influencer on a popular social media site! You decide to start working on making your services more robust. Your plan is to apply a limit to how long your service will process each request. The timeout should default to one second, or it can be provided from the calling service via a header.

You decide to again use the quattro library; it has a useful fail_after context manager that should do the trick.

Since incant supports context managers as dependencies, your path is clear:

from quattro import fail_after

def apply_timeout(timeout: Header = Header("1.0")) -> ContextManager[CancelScope]:
    return fail_after(float(timeout))

However, our usual approach would require refactoring all our handlers to require this dependency. Instead, we will make this a forced dependency, which means incant will run it for all handlers. Since no handler needs the return value of any forced dependency (since they are unaware of them), they are mostly used for side-effects. We change the quickapi decorator thusly:

def quickapi(handler):
    log = logger.bind(handler=handler.__name__)

    prepared = incanter.prepare(handler, forced_deps=[(apply_timeout, "sync")])

    @wraps(handler)
    async def wrapper(**kwargs):
        log.info("Processing")
        return await incanter.aincant(prepared, log=log, **kwargs)

    return wrapper

Complex Rules

Another day, another feature request.

A colleague wants to receive instances of attrs classes, deserialized from JSON in the request body. An example:

@define
class SamplePayload:
    field: int

@app.post("/payload")
@quickapi
async def attrs_handler(payload: SamplePayload, log) -> str:
    log.info("Received payload", payload=repr(payload))
    return "After payload"

They want this to work for any attrs class. You know you can reach for the cattrs library to load an attrs class from JSON, but the dependency hook is a little more complex. Because the dependency hook needs to work for any attrs class, you need to use incanter.register_hook_factory, the most powerful but lowest level hook registration method.

incanter.register_hook_factory is for, like the name says, factories of dependency hooks. It will produce a different dependency hook for each attrs class we encounter, which is what we need.

from attrs import has
from cattrs import structure
from werkzeug.exceptions import BadRequest

def make_attrs_payload_factory(attrs_cls: type):
    async def attrs_payload_factory():
        json = await request.get_json(force=True)
        try:
            return structure(json, attrs_cls)
        except Exception as e:
            raise BadRequest() from e

    return attrs_payload_factory


incanter.register_hook_factory(
    lambda p: has(p.annotation), lambda p: make_attrs_payload_factory(p.annotation)
)

This will also return a 400 status code if the payload cannot be properly loaded.

Because of how incant evaluates dependency rules (newest first), this hook factory needs to be registered before the current_user dependency factory. Otherwise, since our User model is also an attrs class, incant would try loading it from the request body instead of getting it from the current_user dependency factory.

Complex Rules Pt 2: Electric Boogaloo

A colleague wants to receive HTTP headers in their handler. They also want these parameters to be able to have default values. You decide to create a simple string subclass called Header and have your colleague annotate their parameters with it. Their header looks like this:

from typing import NewType

Header = NewType("Header", str)

@app.get("/header")
@quickapi
async def a_header_handler(content_type: Header = Header("none"), log=logger) -> str:
    return f"The header was: {content_type}"

Since each header parameter needs separate logic, you once again reach for hook factories. You remember kebab-case is more commonly used than snake_case for headers, so you apply a small transformation - a parameter named content_type will get the value of the content-type header field.

You write the necessary instructions:

def make_header_factory(name: str, default):
    if default is Parameter.empty:
        return lambda: request.headers[name.replace("_", "-")]
    else:
        return lambda: request.headers.get(name.replace("_", "-"), default)

incanter.register_hook_factory(
    lambda p: p.annotation is Header, lambda p: make_header_factory(p.name, p.default)
)

The complete source code of this mini-project can be found at https://github.com/Tinche/incant/blob/main/tests/quickapi.py.

22.1.0 (2022-09-02)

• Breaking change: due to limitations in autodetecting context managers (both sync and async), context manager dependencies must be explicitly registered by passing is_context_manager="sync" (or async) to the registration functions.

• Injection can be customized on a per-parameter basis by annotating a parameter with Annotated[type, incant.Override(...)].

• Implement support for forced dependencies.

• Sync context managers may now be dependencies.

• incanter.a/incant() now handles unfulfilled parameters with defaults properly.

• Switched to CalVer.

0.3.0 (2022-02-03)

• Properly set the return type annotation when preparing a function.

• A hook override can now force a dependency to be promoted to a parameter (instead of being satisfied) by setting Hook.factory to None.

• Parameters with defaults are now supported for incanter.prepare and incanter.a/invoke.

• incanter.a/incant no longer uses invoke under the hood, to allow greater customization. Previous behavior can be replicated by incant(prepare(fn)).

• Optional arguments of dependencies can now be propagated to final function arguments. Keyword-only arguments of dependencies are still filtered out.

0.2.0 (2022-01-13)

• Introduce incanter.prepare, and make incanter.a/invoke use it. prepare just generates the prepared injection wrapper for a function and returns it, without executing it.

• Remove incanter.parameters, since it’s now equivalent to inspect.signature(incanter.prepare(fn)).parameters.

• Add the ability to pass hook overrides to incanter.prepare, and introduce the incanter.Hook class to make it more usable.

0.1.0 (2022-01-10)

• Initial release.