[QUESTION] Dependency Injection about typer HOT 17 CLOSED

I'm interested in a similar use case, as I'm trying to write a CLI to communicate with another server. The commands require the server address, user name, password, etc. in order to establish the connection.

It would be great, if it was possible to establish the connection (and handle the errors, etc.) in one function and inject the result of this into the single commands, in order to avoid the duplication of the options.

Similar like this:

import typer

app = typer.Typer()


def client(
    host: str = typer.Option(...),
    username: str = typer.Option(...),
    tls: bool = True,
    password: str = typer.Option(..., prompt=True, hide_input=True),

) -> Client:
    client = Client(host)
    if not client.connect(
            username, password, starttls=tls
    ):
        raise typer.Exit(code=1)
    return client


@app.command()
def command1(
    client: Client = typer.Dependency(client)
):
    assert client.is_connected


@app.command()
def command2(
    client: Client = typer.Dependency(client)
):
    pass


if __name__ == "__main__":
    app()

So typer would have to resolve the arguments and options for the dependency, execute that method and pass the value into the command being called. I didn't think this through any further than that, so I don't know what the implications would be.

I'm sure this would also be possible by using meta programming or some fancy decorator that modifies the function signature. But I'd rather not get my hands that dirty.

from typer.

@mawillcockson Thanks for the ideas.

As for making async calls, I created following decorator:

F = TypeVar('F', bound=Callable[..., Any])
...
def async_entrypoint(func: F) -> F:
    @wraps(func)
    def wrapper(*args, **kwargs):
        async def async_wrapper(*args, **kwargs):
            return await func(*args, **kwargs)

        return asyncio.run(async_wrapper(*args, **kwargs))

    return wrapper

Then I annotate the function like this:

@pod.command()
@async_entrypoint
async def run_test(
...

from typer.

@tiangolo I'm wondering why not decouple DI component from FastAPI as a separate package and reuse in both Typer and FastAPI?
I'm currently, building application which has API and CLI interfaces. IMO it is odd that FastAPI has DI implementation but Typer does not. Of course, I can reuse DI from FastAPI, but it does not cover the case when I want only CLI.

from typer.

Ah, yeah, I had seen that style of DI, and that's probably the most traditional way of doing it. But I didn't want to do it that way, as that means there's only one possible provider for a dependency.

So, it was not possible to declare user: User = Depends(valid_user) and then somewhere else user: User = Depends(valid_admin_user), because User would be provided only by one thing.

And also it would mean that dependencies would have to have some specific shape and interface, or quite some extra concepts and API to learn to register providers (including their terminology). So I ended up deciding on the current style.

But I'm glad you found something that suits your needs ✔️ 🤓

from typer.

For example I want to create aiohttp session and use it in various callbacks and the final command and after completion cleanly close it.

Ideally if this could be done in a form of:

async with aiohttp.ClientSession() as session:
    yield session

It would be perfect. Right now I placed async with whenever I need it, but that means there's a separate session for every callback, and this code is being repeated over and over.

I'm actually curious about how you're integrating async callbacks with Typer. Would you have an example or repository on hand, @takeda?

As to handling setup of resources before a command is run, and then teardown after, there's a few approaches supported by click:

Use a resource for all commands

import typer


class ExampleSession:
    def __init__(self, url: str):
        print(f"opening session to '{url}'")
        self.url = url

    def send(self, data: str) -> None:
        print(f"sending '{data}' to '{self.url}'")

    def close(self) -> None:
        print("closing session")


def session_dependency(ctx: typer.Context) -> None:
    "opens a session for every subcommand"
    # Skip creating resource if this is called during completion, or without a
    # subcommand
    if ctx.resilient_parsing or ctx.invoked_subcommand is None:
        return

    ctx.obj = ExampleSession("https://example.com")
    ctx.call_on_close(ctx.obj.close)


app = typer.Typer(callback=session_dependency)


@app.command()
def send(ctx: typer.Context, data: str = typer.Argument(...)) -> None:
    ctx.obj.send(data)


if __name__ == "__main__":
    app()

$ ./example.py send hello
opening session to 'https://example.com'
sending 'hello' to 'https://example.com'
closing session

ctx.with_resource() could also be used instead, to support something that behaves as a context manager.

Use a resource for specific commands

from typing import Optional

import typer
from click import ParamType


class ExampleSession:
    def __init__(self, url: str):
        print(f"opening session to '{url}'")
        self.url = url

    def send(self, data: str) -> None:
        print(f"sending '{data}' to '{self.url}'")

    def close(self) -> None:
        print("closing session")


def session_dependency(ctx: typer.Context) -> ExampleSession:
    "opens a session for decorated subcommands"
    session = ExampleSession("https://example.com")
    ctx.call_on_close(session.close)
    return session


app = typer.Typer()


@app.callback()
def force_send_to_be_subcommand() -> None:
    pass


@app.command()
def send(
    session: Optional[str] = typer.Option(
        None, callback=session_dependency, hidden=True
    ),
    data: str = typer.Argument(...),
) -> None:
    if not session:
        raise typer.Abort("session was not created")

    session.send(data)


if __name__ == "__main__":
    app()

This is lying about the types, but Typer doesn't currently support type annotations that subclass click.ParamType, so I don't know of a better way to do this.

Dependency injection

None of the above methods work particularly well as a way to inject multiple dependencies, and overriding those dependencies during testing.

The callback-based one works better, by using the callback overriding supported by Typer:

import typer
from example import ExampleSession, app
from typer.testing import CliRunner

runner = CliRunner()
TEST_URL = "https://example.test"


def make_test_session(ctx: typer.Context) -> None:
    test_session = ExampleSession(TEST_URL)
    ctx.obj = test_session
    ctx.call_on_close(ctx.obj.close)


def test_send():
    app.callback()(make_test_session)
    test_data = "test"
    result = runner.invoke(app, ["send", test_data])
    assert result.exit_code == 0
    assert f"opening session to '{TEST_URL}'" in result.stdout
    assert f"sending '{test_data}' to '{TEST_URL}'" in result.stdout
    assert "closing session" in result.stdout

But it's still limited by only allowing one callback per Typer() command group.

The latter approach works fine for testing if the commands can be run as isolated functions:

import typer
from example import ExampleSession, send
from typer.testing import CliRunner

runner = CliRunner()
TEST_URL = "https://example.test"


def test_send():
    test_session = ExampleSession(TEST_URL)
    test_data = "test"
    send(session=test_session, data=test_data)

from typer.

Hi,
just landed on this issue when looking for dependency injection with typer. The solution with callbacks seemed a bit tedious to me, so I created a simple example with injector. The nice thing is, it clearly separates our the interface (via typer) and the business logic and providers (or whatever else you have chosen in your app design) via injector.
It's not perfect, but I could imagine it to be a good solution for people simply wanting to have clean dependency injection.

from dataclasses import dataclass

import typer
from injector import inject, Injector


class FancyProvider:
    fancy_value = 24


@inject
@dataclass
class FancyBusiness:
    fancy_provider: FancyProvider

    def fancy_method(self, fancy_argument: int):
        return fancy_argument + self.fancy_provider.fancy_value


def main(my_fancy_argument: int):
    fancy_business = Injector().get(FancyBusiness)
    print(f"Fancy result: {fancy_business.fancy_method(my_fancy_argument)}")


if __name__ == "__main__":
    typer.run(main)

PS: I haven't tried it with async yet :)

from typer.

Hello @tiangolo,

The developer experience is just 👌

You're right, in this case it's not very usefull as we can provide the dependency easily. I was the just wondering if it's possible ?

from typer.

Thank you for your answer, actually it does.

from typer.

So, it was not possible to declare user: User = Depends(valid_user) and then somewhere else user: User = Depends(valid_admin_user), because User would be provided only by one thing.

Multiple instances of a single class (e.g., multiple HTTP clients with different endpoint configurations) is common. More often you can face with dependencies on an interface or some more abstract type due to SOLID principles. Complete DI solutions (e.g., Symfony (PHP), Spring (Java)) give the possibility to declared named dependencies instead of type-based.
Autowiring usually complements named based dependencies declarations. Unfortunately, https://github.com/ivankorobkov/python-inject isn't a complete solution but you always able to declare a new type which is not preferable but solve the issue.

And also it would mean that dependencies would have to have some specific shape and interface, or quite some extra concepts and API to learn to register providers (including their terminology). So I ended up deciding on the current style.

Autowiring usually minimizes the amount of boilerplate code and have a less steep learning curve.

from typer.

Hey @Leletir ! I'm glad the tools are being useful! 🎉

In this use case, what would be the advantage of having the DB session provided by a dependency injection system?

In other words, what would be wrong with something explicit? Creating the DB in the function with:

db = get_db()

like in:

save_app = typer.Typer()


def get_db():
    typer.echo("Initializing database")
    db = VerticaDB(
        host=conf.get_vertica_host(),
        port=conf.get_vertica_port(),
        user=conf.get_vertica_user(),
        password=conf.get_vertica_password(),
        database=conf.get_vertica_database()
    )
    return db


@save_app.command("save")
def ingest_snitch_file(
        path_to_log_file: Path = typer.Option(
            ...,
            exists=True,
            readable=True,
            envvar='PATH_TO_SNITCH_FILE'
        )
):
    """
    Ingest snitch files to Database
    """
    db = get_db()
    snitch = Snitch(
        db=db,
        path_to_snitch=path_to_log_file,
        table_schema=conf.get_vertica_table_schema(),
        table_name=conf.get_vertica_table_name()
    )
    snitch.insert_log_to_database()

from typer.

Thanks for the report @Leletir !

@larsrinn I think you could achieve something very similar by adding those settings to a callback: https://typer.tiangolo.com/tutorial/commands/callback/

That are then shared by each command, probably also using the Context: https://typer.tiangolo.com/tutorial/commands/context/

@Leletir if that answers your question, you can close this issue. ✔️

from typer.

@tiangolo Indeed, I just tested it using a callback and it works (while scrolling through the documentation I just saw callbacks for options and considered this was unsuitable).

For the rather simple application I'm building this solution is fine. But I guess this might get out of hand for more complex stuff, where multiple groups of commands require different setups. Also closing the connection the connection is not that clean now. So I think typer could benefit from a dependency injection system similar to pytest fixtures.

from typer.

@tiangolo seems it is impossible to reuse FastAPI DI for Typer because of the following error:

File ".../typer/main.py", line 586, in get_click_type
    raise RuntimeError(f"Type not yet supported: {annotation}")  # pragma no cover
RuntimeError: Type not yet supported:

from typer.

To resolve this issue I rid off of using FastAPI DI and install https://github.com/ivankorobkov/python-inject to share same DI container between CLI and API.
Currently, it is the only way how to avoid code duplication.

@tiangolo I suggest reviewing python-inject as a replacement for FastAPI DI.

from typer.

The thing is that FastAPI uses the dependency injection system to extract metadata for validation and documentation to be taken from the request.

A dependency function for FastAPI would probably not work for a CLI as it would probably be taking data from a request, e.g. from the body and from cookies. Where would that data come from in the CLI? So, in Typer, there wouldn't be a way to extract that info, so the dependency function would have to be different. And in that case, what benefit would it provide that is not achieved by calling the helper function directly?

from typer.

Where would that data come from in the CLI?

There is a difference between CLI and HTTP input data structures. Still, both have a need to work with some particular dependencies (e.g. services that contain exact logic that can be shared between CLI and HTTP)

So, in Typer, there wouldn't be a way to extract that info, so the dependency function would have to be different.

IMO there should not be a difference between DI module in Typer and FastAPI that why I propose to decouple this part of functionality into a separate package or reuse already existing library.

And in that case, what benefit would it provide that is not achieved by calling the helper function directly?

You probably right. I do not remember have I tried this approach or not. The described error wasn't the only reason why I have migrated to https://github.com/ivankorobkov/python-inject . Unfortunately, with FastAPI DI I have to describe all dependencies manually which isn't productive. I was looking for a library that solves DI task using auto wiring approach which utilizes already existing types hints.

from typer.

@tiangolo I have somewhat related but slightly different question. Is there a way to create an object at the beginning of the lifecycle of a command and terminate at the end?

For example I want to create aiohttp session and use it in various callbacks and the final command and after completion cleanly close it.

Ideally if this could be done in a form of:

async with aiohttp.ClientSession() as session:
    yield session

It would be perfect. Right now I placed async with whenever I need it, but that means there's a separate session for every callback, and this code is being repeated over and over.

from typer.