0

I want to implement my own Dependency Injection like Fastapi Depends() do actually without using external package or framework. What will be the approach? Code example will be helpful for me. Thanks in advance.

from typing import Callable, Optional, Any

class Depends:
    def __init__(self, dependencies= Optional[Callable[..., Any]]):
        self.dependencies = dependencies
        
        
def get_db():
    pass

    
def get_token():
    pass

def get_current_user(db= Depends(get_db),  token= Depends(get_token)):
    pass
  • So which features do you want your dependency injection library to support? You can do dependency injection with something as simple as a dict by itself, but you're probably looking for more than that - do you want to support single function calls like FastAPI dependencies? Multi-level dependencies in that manner? Do you want to support caching? - i.e. do you want to read through FastAPIs source to find out how it works there instead? – MatsLindh Oct 22 '22 at 21:17
  • @MatsLindh I think you understand what i actually want to implement. I want single function call like Fastapi. – Mahbub Rahman Oct 23 '22 at 07:03
  • 1
    You can probably implement it through a small class that implements callable and takes a Callable as an argument. When it gets called, you proxy that call to the callable that have been added; that way you can wrap any callable and only call them when they're actually used. You can then expand this further by going through the arguments for the function you're calling and seeing if they're an instances of your dependency handler, and then check if you've already resolved that dependency to avoid calling it again (the `cache` part). – MatsLindh Oct 23 '22 at 18:20
  • FastAPI does quite a bit of magic to make it work with the request cycle, (I'm halfway guessing this part, but I think this is fairly accurate) and since any method that has their dependencies evaluated is wrapped in another function (i.e. decorated with a route), this function is what resolves the dependencies for a given situation (and can make each request have its own dependant values) . The heavy lifting is down by the decorator function, which provides values to all the underlying functions. – MatsLindh Oct 23 '22 at 18:39

2 Answers2

1

A starting point could be something like this, where we create a decorator that allows us to preempt any calls to the function and resolve any dependencies.

from typing import Dict, Callable, Any
from functools import wraps
import inspect


# Our decorator which inspects the function and resolves any
# dependencies when called
def resolve_dependencies(func):
    # based on https://stackoverflow.com/a/69170441/
    f_sig = inspect.signature(func)

    @wraps(func)
    def resolve_nice_to_have(*args, **kwargs):
        bound = f_sig.bind(*args, **kwargs)
        bound.apply_defaults()

        for k, arg in bound.arguments.items():
            if type(arg) == ItWouldBeNice:
                bound.arguments[k] = arg()

        return func(*bound.args, **bound.kwargs)

    return resolve_nice_to_have


# Our actual dependency wrapper, with a simple cache to avoid
# invocating an already resolved dependency.
# Slightly friendlier named than actually depending on something.
class ItWouldBeNice:
    cache: Dict[Callable, Any] = {}

    def __init__(self, dependency: Callable):
        self.dependency = dependency

    def __call__(self) -> Any:
        if self.dependency in ItWouldBeNice.cache:
            return ItWouldBeNice.cache[self.dependency]

        result = self.dependency()
        ItWouldBeNice.cache[self.dependency] = result
        return result

Example of usage:

from iwant import ItWouldBeNice, resolve_dependencies


def late_eval():
    print("late was called")
    return "static string"


@resolve_dependencies
def i_want_it(s: str = ItWouldBeNice(late_eval)):
    print(s)


@resolve_dependencies
def i_want_it_again(s: str = ItWouldBeNice(late_eval)):
    print(s)


i_want_it()
i_want_it_again()

This doesn't support hierarchical dependencies etc., but should at least illustrate a concept you could apply to do something similar.

MatsLindh
  • 49,529
  • 4
  • 53
  • 84
-1

You can do something like this?

async def get_db(db_con=Depends(get_db_con)) -> AsyncIterable[Session]:
    session = Session(bind=db_con)
    try:
        yield session
    finally:
        session.close()

The get_db_con function could return the initialised database (or raise a connection error).

Hope this helps

cheers.