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.
- 1You register itAt startup, you tell risa which object a
Pricesparameter gets. - 2The handler names itIt takes a parameter,
prices: Prices. - 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
Shop
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
Apple: 3 coins
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.
- 1The clickrisa opens a
COMPONENTcontainer and putsctxand the interaction in it. - 2
load()and the handlerFor each parameter, the container hands over what it holds, or makes it. - 3The end of the clickThe container closes and tears down what it made.
- 4The next clickIt gets a new container. The
DEFAULTone stays open.
Two rules follow from this:
- A container only looks outward. A dependency in
DEFAULTcan't name one inCOMPONENT, and a click that needs it fails withDependencyNotSatisfiableException. - A registry freezes while a container made from it is open. The
DEFAULTcontainer opens at the first click orload()and stays open, so a laterregister_valueraises linkd'sRegistryFrozenException. 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
An apple costs 3 coins.
After a click on Buy
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.