Skip to content

Dependency injection

A handler often needs something it doesn't make itself, such as your database. Register that thing once, name it as a parameter, and risa passes it in on every click.

What it is

A dependency is anything your code needs but doesn't make itself: a database, an HTTP session, your config. A function can get one in two ways. It can open the thing itself:

async def price_of(item: str) -> int:
    database = await connect("postgres://localhost/shop")
    return await database.price(item)

That opens a new connection on every call, and each function that needs the database has to know how to open it. Or the function can name what it needs and leave the rest to whoever calls it:

async def price_of(item: str, database: Database) -> int:
    return await database.price(item)

This is dependency injection: the function says what it needs, and its caller passes it in. Every call can share one database, and only one place in your bot knows how to open it.

A handler works the same way, except that you never call it. A click comes in, and risa calls it. So you tell risa at startup what to pass, and the handler names what it wants. risa does this through linkd, the library lightbulb uses as well.

  1. 1You register itAt startup, you tell risa which object a Prices parameter gets.
  2. 2The handler names itIt takes a parameter, prices: Prices.
  3. 3risa passes it inOn each click, risa calls the handler with that object.

A view's field can't hold your database: fields get saved from click to click, and @risa.register raises ViewDeclarationError on a type it can't save. A global variable can, until you have something that opens only once the bot is running, such as a connection pool. The global stays empty until then, and each handler has to check it.

Register it

This shop shows a price when someone clicks an item. Prices stands in for your database, and you can find it in the full file.

In Discord

Risa BotAPPToday at 12:00

Shop

ApplePear

main() calls this once, before the bot starts:

def add_dependencies(client: risa.Client) -> None:
    registry = client.di.registry_for(risa.Contexts.DEFAULT)
    registry.register_value(Prices, Prices())

register_value(Prices, Prices()) says that a parameter annotated Prices gets this one object. How it works explains the line above it.

Name it as a parameter

A click on an item runs show, which names what it needs.

@risa.handler
async def show(
    self,
    ctx: risa.ComponentContext,
    item: str,
    prices: Prices,
    client: risa.Client,
) -> None:
    offer = Offer(item=item, price=await prices.of(item))
    await ctx.respond(**await client.build(offer), ephemeral=True)
Parameter Filled by
ctx risa, by position
item: str The wire argument bound to the button
prices: Prices, client: risa.Client Injection, going by the annotation, so any name will do

risa tells them apart by type. str is a type an argument can have, so item comes from the button. Prices is not, so prices is injected.

If nothing is registered for a parameter, the handler does not run: risa logs linkd's DependencyNotSatisfiableException, and the click fails in Discord.

Build another view

show also names client, which you never registered, because risa registers it for you. A handler needs it for client.build, since the client is a local variable of main(). show builds an Offer and sends that as its answer. This is the other view, and main() adds both to the client.

@risa.register(name="offer")
class Offer(risa.View):
    item: str
    price: int

    def render(self) -> ui.Layout:
        return ui.Container(
            ui.TextDisplay(f"{self.item.title()}: {self.price} coins"),
            ui.Row(ui.Button(self.buy, label="Buy")),
        )

    @risa.handler
    async def buy(self, ctx: risa.ComponentContext) -> None:
        await ctx.edit(ui.TextDisplay(f"You bought the {self.item}."))

The answer to a click on Apple, which only that person sees

Risa BotAPPToday at 12:00

Apple: 3 coins

Buy
Full file
"""A shop whose handler takes a wire argument, your database and the client."""

from __future__ import annotations

import asyncio
import os

import hikari

import risa
from risa import ui

_PRICES = {"apple": 3, "pear": 4}


class Prices:
    async def of(self, item: str) -> int:
        await asyncio.sleep(0.1)
        return _PRICES[item]


@risa.register(name="offer")
class Offer(risa.View):
    item: str
    price: int

    def render(self) -> ui.Layout:
        return ui.Container(
            ui.TextDisplay(f"{self.item.title()}: {self.price} coins"),
            ui.Row(ui.Button(self.buy, label="Buy")),
        )

    @risa.handler
    async def buy(self, ctx: risa.ComponentContext) -> None:
        await ctx.edit(ui.TextDisplay(f"You bought the {self.item}."))


@risa.register(name="shop")
class Shop(risa.View):
    def render(self) -> ui.Layout:
        buttons = [
            ui.Button(
                risa.bind(self.show, item),
                label=item.title(),
            )
            for item in _PRICES
        ]
        return ui.Container(ui.TextDisplay("## Shop"), ui.Row(*buttons))

    @risa.handler
    async def show(
        self,
        ctx: risa.ComponentContext,
        item: str,
        prices: Prices,
        client: risa.Client,
    ) -> None:
        offer = Offer(item=item, price=await prices.of(item))
        await ctx.respond(**await client.build(offer), ephemeral=True)


def add_dependencies(client: risa.Client) -> None:
    registry = client.di.registry_for(risa.Contexts.DEFAULT)
    registry.register_value(Prices, Prices())


def main() -> None:
    bot = hikari.GatewayBot(
        os.environ["DISCORD_TOKEN"],
        intents=hikari.Intents.ALL_UNPRIVILEGED | hikari.Intents.MESSAGE_CONTENT,
    )
    client = risa.client_from_app(bot)
    add_dependencies(client)
    client.add_view(Shop)
    client.add_view(Offer)

    @bot.listen()
    async def on_message(event: hikari.GuildMessageCreateEvent) -> None:
        if event.content == "!shop":
            rendered = await client.build(Shop())
            await rendered.send_to(bot.rest, event.channel_id)

    bot.run()


if __name__ == "__main__":
    main()

That covers most bots. The rest of this page explains how injection works and what else it can do.

What risa registers

risa registers these itself, so you can name them in any handler without registering anything.

For the whole run What it gets
risa.Client The risa client, for client.build. risa.GatewayEnabledClient is not registered, so annotate it this way
hikari.api.RESTClient The bot's REST client, for calling Discord beyond the answer to this click
hikari.GatewayBot On a gateway bot, the bot, with its cache in bot.cache
hikari.api.EventManager On a gateway bot, its event manager
hikari.RESTBot On a REST bot, the bot
hikari.api.InteractionServer On a REST bot, its interaction server
For each click What it gets
risa.ComponentContext The click, the same object as ctx. A factory or a function further down uses this to reach it
hikari.ComponentInteraction The interaction behind the click, the same object as ctx.interaction

This handler uses the REST client to post in a log channel as well as answering the click:

@risa.handler
async def report(
    self,
    ctx: risa.ComponentContext,
    rest: hikari.api.RESTClient,
) -> None:
    await rest.create_message(_LOG_CHANNEL, f"{ctx.user.username} reported a problem.")
    await ctx.respond("Thanks, we'll take a look.", ephemeral=True)

How it works

Three things work together.

Term What it is
Context A lifetime with a name. risa.Contexts.DEFAULT lasts for the bot's whole run, and risa.Contexts.COMPONENT lasts for one click
Registry The list of what a context can provide and how to make it. register_value adds to this list
Container A box of live objects, made from a registry while its context is open. It makes each object the first time something asks for it, keeps it, and tears it down when the box closes

So client.di.registry_for(risa.Contexts.DEFAULT) reads as "the list for the whole run", and client.di is linkd's manager, which holds every list and box. There is one more context, risa.Contexts.MODAL, which waits for modals.

The boxes sit one inside the other. The objects below are the ones from the shop and the stall further down.

DEFAULT one container, open for the bot's whole run

risa.Client hikari.api.RESTClient Prices Database

COMPONENT a new container for each click

risa.ComponentContext hikari.ComponentInteraction Wallet

A handler runs here. It looks in this box first, then in the one around it.

  1. 1The clickrisa opens a COMPONENT container and puts ctx and the interaction in it.
  2. 2load() and the handlerFor each parameter, the container hands over what it holds, or makes it.
  3. 3The end of the clickThe container closes and tears down what it made.
  4. 4The next clickIt gets a new container. The DEFAULT one stays open.

Two rules follow from this:

  • A container only looks outward. A dependency in DEFAULT can't name one in COMPONENT, and a click that needs it fails with DependencyNotSatisfiableException.
  • A registry freezes while a container made from it is open. The DEFAULT container opens at the first click or load() and stays open, so a later register_value raises linkd's RegistryFrozenException. Register before the bot starts.

Values and factories

register_value takes an object you already have. register_factory takes a function, and the container calls it the first time something names the type. Use it for what has to open inside the running event loop, such as a connection pool.

async def open_database(config: Config) -> Database:
    return await Database.connect(config.database_url)


def add_database(client: risa.Client) -> None:
    registry = client.di.registry_for(risa.Contexts.DEFAULT)
    registry.register_value(Config, Config())
    registry.register_factory(Database, open_database, teardown=Database.close)

A factory can be async, and its parameters are injected too, so open_database names Config and gets it from the same registry. If a factory raises, the handler does not run: risa logs DependencyNotSatisfiableException, and the click fails. Cleanup covers teardown.

A dependency for each click

A wallet belongs to whoever clicked, so it can't exist before the click. Register it in COMPONENT, and each click gets its own.

async def open_wallet(ctx: risa.ComponentContext, database: Database) -> Wallet:
    coins = await database.coins_of(ctx.user.id)
    return Wallet(database, ctx.user.id, coins)


def add_wallet(client: risa.Client) -> None:
    registry = client.di.registry_for(risa.Contexts.COMPONENT)
    registry.register_factory(Wallet, open_wallet)

open_wallet names the click, which risa puts in the COMPONENT container, and Database, which the container finds in the DEFAULT one around it. A handler names Wallet like anything else.

@risa.handler
async def buy(self, ctx: risa.ComponentContext, wallet: Wallet) -> None:
    if wallet.coins < self.price:
        await ctx.respond("You can't afford that.", ephemeral=True)
        return
    await wallet.pay(self.price)
    await ctx.respond(f"You have {wallet.coins} coins left.", ephemeral=True)

In Discord

Risa BotAPPToday at 12:00

An apple costs 3 coins.

Buy

After a click on Buy

Risa BotAPPToday at 12:00

You have 7 coins left.

Only you can see this·Dismiss message

The container makes a wallet at most once per click, so load() and the handler of one click share it, and a handler that doesn't name it never opens one.

client.build runs load() in DEFAULT alone, even inside a handler. A load() that names Wallet makes it raise DependencyNotSatisfiableException, and one that names Wallet | None gets None there.

Full file
"""A stall whose wallets open for each click, from a database that opens once."""

from __future__ import annotations

import os

import hikari

import risa
from risa import ui


class Config:
    database_url = "postgres://localhost/shop"


class Database:
    def __init__(self, url: str) -> None:
        self.url = url
        self.coins: dict[hikari.Snowflake, int] = {}

    @classmethod
    async def connect(cls, url: str) -> Database:
        return cls(url)

    async def close(self) -> None:
        self.coins.clear()

    async def coins_of(self, user: hikari.Snowflake) -> int:
        return self.coins.get(user, 10)

    async def set_coins(self, user: hikari.Snowflake, coins: int) -> None:
        self.coins[user] = coins


class Wallet:
    def __init__(self, database: Database, user: hikari.Snowflake, coins: int) -> None:
        self.database = database
        self.user = user
        self.coins = coins

    async def pay(self, price: int) -> None:
        self.coins -= price
        await self.database.set_coins(self.user, self.coins)


async def open_database(config: Config) -> Database:
    return await Database.connect(config.database_url)


def add_database(client: risa.Client) -> None:
    registry = client.di.registry_for(risa.Contexts.DEFAULT)
    registry.register_value(Config, Config())
    registry.register_factory(Database, open_database, teardown=Database.close)


async def open_wallet(ctx: risa.ComponentContext, database: Database) -> Wallet:
    coins = await database.coins_of(ctx.user.id)
    return Wallet(database, ctx.user.id, coins)


def add_wallet(client: risa.Client) -> None:
    registry = client.di.registry_for(risa.Contexts.COMPONENT)
    registry.register_factory(Wallet, open_wallet)


@risa.register(name="stall")
class Stall(risa.View):
    price: int = 3

    def render(self) -> ui.Layout:
        return ui.Container(
            ui.TextDisplay(f"An apple costs {self.price} coins."),
            ui.Row(ui.Button(self.buy, label="Buy")),
        )

    @risa.handler
    async def buy(self, ctx: risa.ComponentContext, wallet: Wallet) -> None:
        if wallet.coins < self.price:
            await ctx.respond("You can't afford that.", ephemeral=True)
            return
        await wallet.pay(self.price)
        await ctx.respond(f"You have {wallet.coins} coins left.", ephemeral=True)


def main() -> None:
    bot = hikari.GatewayBot(
        os.environ["DISCORD_TOKEN"],
        intents=hikari.Intents.ALL_UNPRIVILEGED | hikari.Intents.MESSAGE_CONTENT,
    )
    client = risa.client_from_app(bot)
    add_database(client)
    add_wallet(client)
    client.add_view(Stall)

    @bot.listen()
    async def on_message(event: hikari.GuildMessageCreateEvent) -> None:
        if event.content == "!stall":
            rendered = await client.build(Stall())
            await rendered.send_to(bot.rest, event.channel_id)

    @bot.listen()
    async def on_stopping(_: hikari.StoppingEvent) -> None:
        await client.di.close()


    bot.run()


if __name__ == "__main__":
    main()

Overriding a dependency

Register a type in COMPONENT that DEFAULT has too, and a click gets the COMPONENT one. A dependency made in DEFAULT keeps using the DEFAULT one, because a container only looks outward.

Adding to an open container

A frozen registry takes nothing new, but an open container does. container.add_value and container.add_factory add to that one container, and what they add goes when it closes. linkd calls these ephemeral dependencies, and risa adds ctx this way. A handler reaches its container by naming it, and a function further down then gets what it added.

@risa.handler
async def flag(
    self,
    ctx: risa.ComponentContext,
    container: risa.ComponentContainer,
) -> None:
    container.add_value(Reason, Reason(f"Flagged by {ctx.user.username}."))
    await record()
    await ctx.respond("Flagged.", ephemeral=True)
Container What it gets
risa.ComponentContainer The click's container
linkd.RootContainer The DEFAULT container
linkd.Container The innermost open container, which in a handler is the click's

Two of one type

A type has one registration. For two databases, make a class for each, and name those.

class Primary(Database):
    pass


class Replica(Database):
    pass


def add_databases(client: risa.Client) -> None:
    registry = client.di.registry_for(risa.Contexts.DEFAULT)
    registry.register_value(Primary, Primary("primary"))
    registry.register_value(Replica, Replica("replica"))
@risa.handler
async def vote(
    self,
    ctx: risa.ComponentContext,
    primary: Primary,
    replica: Replica,
) -> None:
    await primary.add_vote(ctx.user.id)
    total = await replica.count_votes()
    await ctx.respond(f"{total} votes so far.", ephemeral=True)

If a library makes the object for you, so that a subclass won't do, wrap it in a small class of your own. linkd also accepts a typing.NewType as the type, but pyright rejects it in register_value.

Optional dependencies

Annotate a parameter Metrics | None, and it gets None when nothing is registered.

@risa.handler
async def like(self, ctx: risa.ComponentContext, metrics: Metrics | None) -> None:
    if metrics is not None:
        metrics.count("likes")
    await ctx.respond("Thanks!", ephemeral=True)

A union falls back the same way: Memory | Disk gets the first one registered. linkd.Try[Memory] | Disk also falls back when making Memory fails, where a plain union would fail the click.

Where injection works

Handlers, load(), on_state_missing, on_outdated, error handlers and factories have it without a decorator. For a function of your own, add @linkd.inject. render() takes no injected parameters, so load() reads what it draws.

A parameter is injected if it has an annotation and either no default or the default risa.INJECTED. Any other default stays as it is. *args, **kwargs and positional-only parameters are left alone. A parameter with no annotation gets nothing, so the click fails with a TypeError.

risa.INJECTED tells your type checker that nobody has to pass the parameter. on_outdated, on_state_missing and a function of your own that you call with no arguments need it. In a handler it also marks a wire type as injected: limit: int = risa.INJECTED.

Injection context

An injected function needs an open context when it runs. risa opens DEFAULT and COMPONENT around load(), the handler, the hooks and the error handlers of a click, and DEFAULT around the load() in client.build. A function called from inside those has them too, so you rarely need to think about this.

Anywhere else, such as in a hikari listener, open the context yourself, or the call raises DependencyNotSatisfiableException.

@linkd.inject
async def post_votes(
    rest: hikari.api.RESTClient = risa.INJECTED,
    replica: Replica = risa.INJECTED,
) -> None:
    total = await replica.count_votes()
    await rest.create_message(_LOG_CHANNEL, f"{total} votes so far.")


def post_on_start(bot: hikari.GatewayBot, client: risa.Client) -> None:
    @bot.listen()
    async def on_started(_: hikari.StartedEvent) -> None:
        async with client.di.enter_context(risa.Contexts.DEFAULT):
            await post_votes()
Full file
"""A feedback message that shows what else injection can do."""

from __future__ import annotations

import dataclasses
import os

import hikari
import linkd

import risa
from risa import ui

_LOG_CHANNEL = hikari.Snowflake(123456789)
_VOTES: list[hikari.Snowflake] = []


class Database:
    def __init__(self, name: str) -> None:
        self.name = name

    async def add_vote(self, user: hikari.Snowflake) -> None:
        _VOTES.append(user)

    async def count_votes(self) -> int:
        return len(_VOTES)


class Metrics:
    def __init__(self) -> None:
        self.counts: dict[str, int] = {}

    def count(self, name: str) -> None:
        self.counts[name] = self.counts.get(name, 0) + 1


@dataclasses.dataclass
class Reason:
    text: str


class Primary(Database):
    pass


class Replica(Database):
    pass


def add_databases(client: risa.Client) -> None:
    registry = client.di.registry_for(risa.Contexts.DEFAULT)
    registry.register_value(Primary, Primary("primary"))
    registry.register_value(Replica, Replica("replica"))


@linkd.inject
async def record(
    reason: Reason = risa.INJECTED,
    rest: hikari.api.RESTClient = risa.INJECTED,
) -> None:
    await rest.create_message(_LOG_CHANNEL, reason.text)


@risa.register(name="feedback")
class Feedback(risa.View):
    def render(self) -> ui.Layout:
        return ui.Container(
            ui.TextDisplay("Did this help?"),
            ui.Row(
                ui.Button(self.vote, label="Yes"),
                ui.Button(self.like, label="Love it"),
                ui.Button(self.report, label="Report a problem"),
                ui.Button(self.flag, label="Flag"),
            ),
        )

    @risa.handler
    async def vote(
        self,
        ctx: risa.ComponentContext,
        primary: Primary,
        replica: Replica,
    ) -> None:
        await primary.add_vote(ctx.user.id)
        total = await replica.count_votes()
        await ctx.respond(f"{total} votes so far.", ephemeral=True)

    @risa.handler
    async def like(self, ctx: risa.ComponentContext, metrics: Metrics | None) -> None:
        if metrics is not None:
            metrics.count("likes")
        await ctx.respond("Thanks!", ephemeral=True)

    @risa.handler
    async def report(
        self,
        ctx: risa.ComponentContext,
        rest: hikari.api.RESTClient,
    ) -> None:
        await rest.create_message(_LOG_CHANNEL, f"{ctx.user.username} reported a problem.")
        await ctx.respond("Thanks, we'll take a look.", ephemeral=True)

    @risa.handler
    async def flag(
        self,
        ctx: risa.ComponentContext,
        container: risa.ComponentContainer,
    ) -> None:
        container.add_value(Reason, Reason(f"Flagged by {ctx.user.username}."))
        await record()
        await ctx.respond("Flagged.", ephemeral=True)


@linkd.inject
async def post_votes(
    rest: hikari.api.RESTClient = risa.INJECTED,
    replica: Replica = risa.INJECTED,
) -> None:
    total = await replica.count_votes()
    await rest.create_message(_LOG_CHANNEL, f"{total} votes so far.")


def post_on_start(bot: hikari.GatewayBot, client: risa.Client) -> None:
    @bot.listen()
    async def on_started(_: hikari.StartedEvent) -> None:
        async with client.di.enter_context(risa.Contexts.DEFAULT):
            await post_votes()


def main() -> None:
    bot = hikari.GatewayBot(
        os.environ["DISCORD_TOKEN"],
        intents=hikari.Intents.ALL_UNPRIVILEGED | hikari.Intents.MESSAGE_CONTENT,
    )
    client = risa.client_from_app(bot)
    add_databases(client)
    client.add_view(Feedback)
    post_on_start(bot, client)

    @bot.listen()
    async def on_message(event: hikari.GuildMessageCreateEvent) -> None:
        if event.content == "!feedback":
            rendered = await client.build(Feedback())
            await rendered.send_to(bot.rest, event.channel_id)

    bot.run()


if __name__ == "__main__":
    main()

Cleanup

Give a dependency a teardown, a function that takes the dependency, as open_database above does with Database.close. It runs when the dependency's container closes, and only if the container made it.

A COMPONENT container closes at the end of each click, even if the handler raised. The DEFAULT container stays open until you close it, and risa never does, so call client.di.close() when the bot stops.

@bot.listen()
async def on_stopping(_: hikari.StoppingEvent) -> None:
    await client.di.close()
async def close_dependencies(_: hikari.RESTBot) -> None:
    await client.di.close()

bot.add_shutdown_callback(close_dependencies)

With lightbulb, lightbulb_client.stop closes the shared manager, so bot.subscribe(hikari.StoppingEvent, lightbulb_client.stop) is enough.

Share one manager with lightbulb

risa.client_from_lightbulb makes a client on lightbulb's manager, so what you register with lightbulb reaches your handlers too. lightbulb registers the hikari types above itself.

registry = lightbulb_client.di.registry_for(lightbulb.di.Contexts.DEFAULT)
registry.register_value(Prices, Prices())
risa_client = risa.client_from_lightbulb(lightbulb_client)
risa_client.add_view(Shop)
lightbulb_client.register(ShopCommand)

A command gets the risa client the same way:

class ShopCommand(
    lightbulb.SlashCommand,
    name="shop",
    description="Open the shop.",
):
    @lightbulb.invoke
    async def invoke(self, ctx: lightbulb.Context, client: risa.Client) -> None:
        await ctx.respond(**await client.build(Shop()))

risa.Contexts.DEFAULT and lightbulb's DEFAULT are the same context, linkd's root. A click opens risa's COMPONENT and a command opens lightbulb's COMMAND, and neither sees what the other registers.

Full file
"""The shop on lightbulb's manager, so one registration reaches both."""

from __future__ import annotations

import asyncio
import os

import hikari
import lightbulb

import risa
from risa import ui

_PRICES = {"apple": 3, "pear": 4}


class Prices:
    async def of(self, item: str) -> int:
        await asyncio.sleep(0.1)
        return _PRICES[item]


@risa.register(name="shop")
class Shop(risa.View):
    def render(self) -> ui.Layout:
        buttons = [
            ui.Button(
                risa.bind(self.show, item),
                label=item.title(),
            )
            for item in _PRICES
        ]
        return ui.Container(ui.TextDisplay("## Shop"), ui.Row(*buttons))

    @risa.handler
    async def show(
        self,
        ctx: risa.ComponentContext,
        item: str,
        prices: Prices,
    ) -> None:
        price = await prices.of(item)
        await ctx.respond(f"{item.title()}: {price} coins", ephemeral=True)


class ShopCommand(
    lightbulb.SlashCommand,
    name="shop",
    description="Open the shop.",
):
    @lightbulb.invoke
    async def invoke(self, ctx: lightbulb.Context, client: risa.Client) -> None:
        await ctx.respond(**await client.build(Shop()))


def main() -> None:
    bot = hikari.GatewayBot(os.environ["DISCORD_TOKEN"])
    lightbulb_client = lightbulb.client_from_app(bot)
    bot.subscribe(hikari.StartingEvent, lightbulb_client.start)

    registry = lightbulb_client.di.registry_for(lightbulb.di.Contexts.DEFAULT)
    registry.register_value(Prices, Prices())
    risa_client = risa.client_from_lightbulb(lightbulb_client)
    risa_client.add_view(Shop)
    lightbulb_client.register(ShopCommand)

    bot.run()


if __name__ == "__main__":
    main()

Turning it off

Set the environment variable LINKD_DI_DISABLED to true to turn linkd off. risa passes ctx and the wire arguments itself, so a handler that takes only those keeps working. A handler that names anything else fails with a TypeError.

Types have to exist at runtime

linkd reads annotations while the bot runs, so a dependency's type has to be importable then. A type imported only under if typing.TYPE_CHECKING: makes @risa.register raise HandlerSignatureError for a handler, and NameError for a load() or a factory. Keep that import outside the block.