Transient or Scoped dependency support

[sparky2708]

Is there a way to implement transient or scoped dependency? I am thinking of a way to create services that would only exist for the life of the api request in FastApi. E.g. let’s say you get a message with 2 different sections:

[
{trees: [ {tree: {id: 1, desc: “pine”}, …]
{dogs: [ {dog: {id: 1, name: “potato”}, …]
]

When I receive this message I want to use a TreeProvider and DogProvider that source their data from the above message and can be injected just for processing that request and not for other requests.

@inject() #message will be injected
def class DogProvider(IDogProvider):
     def __init__(self, message):
          …

     def get_dogs() -> Dog:
          …

@inject() #message will be injected
def class TreeProvider(ITreeProvider):
     def __init__(self, message)
          …

     def get_trees() -> Tree:
          …

This is similar to a C# scoped or transient registration.

[dkraczkowski]

@sparky2708 I think it can be done, my only concern is that I don't think I will be able to find time to work on it. The way I use kink in my professional work differs slightly, and since I don't need this feature, its development might not be prioritized from my side. However, I'm happy to collaborate with you on it and provide my feedback.

[sparky2708]

For process-scope I have figured out how to achieve it:

import threading

class MyController:
     def __init__():
           self.context = threading.local()
           di[IMyService] = lambda _: self.context.my_service

    def process_api_request(self, request):
           self.context.my_service = inject()(MyService)(date=request.date)
           . . .

[sparky2708]

Ideally I’d like to see it like the following - will try to work on it when I have some free time and create a PR:

class MyController:    
    def process_api_request(self, request):
           with di.create_scope() as scope:
                scope[IMyService] = inject()(MyService)(date=request.date)
                . . .
           #scope is destroyed 
           . . .
 

[jalelegenda]

I believe all of these are already covered. Using your example:

from kink import inject, di


class IScoped:
    def __enter__(self):
        return self

    def __exit__(self, *_):
        print("I'm out")

class Scoped(IScoped):
    pass


class ISingleton:
    def __call__(self):
        print("Yo I'm called")
        print(id(self))
        return self


    def hey(self):
        print(id(self))


class Singleton(ISingleton):
    pass

class ITransient:
    def method(self):
        print("I'm dead")


class Transient(ITransient):
    pass


di[type[IScoped]] = lambda _: Scoped
di[ISingleton] = Singleton()
di[ITransient] = lambda _: Transient()



@inject
def test_scoped(manager: type[IScoped]):
    with manager():
        print("Doing stuff...")


@inject
def test_singleton(singleton: ISingleton):
    print("Testing singleton")
    singleton.hey()


@inject
def test_singleton_again(singleton: ISingleton):
    print("Testing singleton again")
    singleton.hey()


@inject
def test_transient(transient: ITransient):
    print("Testing transient")
    transient.method()


test_scoped()
test_singleton()
test_singleton_again()
test_transient()

Also, in .NET, scoped is a special case referring to the duration of the entire request/response cycle inside a web framework. I'm sure you could achieve that somehow, didn't really give it too much though.

[sparky2708]

Thank you for that. The scoped isn't really right. What if you want to create multiple services that are all valid in 1 scope? Essentially you would want a Container() to be created and destroyed in the same scope and this container would supersede the main container during resolution of an object. i.e. we would first check the scoped container for any scoped registrations.

[av-guy]

Hi @sparky2708,

Can you show me a concrete example of how you would handle this in FastAPI? Would you use the context manager within the body of function designed to handle the request?

For example:

@app.get("/endpoint")
async def get_endpoint(payload: YourDataTypeHere):
    with di.create_scope() as scope:
        ...

If you could give me a concrete example of what you're looking for, I think It would help me to better contextualize a potential solution.