Pass query bindings as dependencies

[tyzhnenko]

Idea

Pass query bindings to DI classes, like in the example below.

class DataStorageFabric:
    def __init__(self, x_storage: FromHeader[str] = FromHeader['default']):
        if 'slow' == x_storage:
            self._storage_type = 'slow'
        elif 'fast' == x_storage:
            self._storage_type = 'fast'
        elif 'default' == x_storage:
            self._storage_type = 'default'

    def storage_engine(self):
        if self._storage_type == 'default':
            return DefatulStorage()
        elif self._storage_type == 'slow':
            return SlowStorage()
        elif self._storage_type == 'fast':
            return FastStorage()


class MediaProcessor:
     def __init__(self, storage_fabric: DataStorageFabric):
         self.storage = storage_fabric.storage_engine()

     def make_image_better(self, image_file):
         image_object = self.save(image_file)
         # some code that make the image great again :)

     def save_image(self, image_file):
         return self.storage.save(image_file)


app = Application()

app.services.add_exact_scoped(MediaProcessor)
app.services.add_exact_scoped(DataStorageFabric)


@app.route.post("/improve-image")
def save_image(file: FromFiles, processor: MediaProcessor):
    processor.make_image_better(file)

Current Situation

BlackSheep gives users a good flexible DI solution. It really helps keep code more structured and readable. But in case a user needs to pass some data from a query, they should do everything manually in a request handler. Get data from bindings and pass it to an injected class as a parameter for a method.

Possible Solution

TBD

[RobertoPrevato]

Hi @tyzhnenko
This is already supported - let me prepare an example. 😄 I write ASAP.

[RobertoPrevato]

Yes, using a factory to obtain the request works, I just prepared the example below. I can explain the rationale for this (below the example).

from blacksheep import Application
from blacksheep.messages import Request
from blacksheep.server.di import dependency_injection_middleware


class ServiceNeedingRequest:
    request: Request


app = Application()


def request_factory(context) -> Request:
    return context.scoped_services["__request__"]


app.middlewares.append(dependency_injection_middleware)
app.services.add_scoped(ServiceNeedingRequest)
app.services.add_scoped_by_factory(request_factory)


@app.route("/")
def example(service: ServiceNeedingRequest):
    print(service.request)
    return "OK"


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, port=44555, lifespan="on")

rodi only raises exception when it knows it would not be able to instantiate a service. In this case we register a factory (a function that explicitly resolves the service as we want). The service in this case cannot be obtained through composition, it is a dynamic value available at runtime for a specific web request event, so it's fine, we make sure the request is available as a scoped service in the container:

• depedency_injection_middleware makes sure that a single DI resolution scope is created for each request. I don´t activate this by default because it incurs a small performance fee and not everybody might need it. This middleware also makes the web request accessible for the resolution scope.
• then I register a factory that returns the request object from the resolution context
• now I can inject scoped services like the User obtained from an authentication cookie, or an Authorization: Bearer token.

This might seem silly, but in fact it's very similar to what you get in other web frameworks that support the same, and I swear I was even myself surprised of how my implementation ended up being similar to others. I guess that's because there are not many ways to resolve this case anyway. In practice you need a context for each web request and being able to bind a service reading that context.

[RobertoPrevato]

In v2 I added the method register_http_context to simplify this action for end users. The most common scenario for enabling this, is as I mentioned having the User object injected through dependency injection.

Finally, I recommend to keep your business logic / domain classes abstracted from HTTP stuff. In other words, I recommend to not inject things like HTTP Header, but using a dedicated class, abstracted from a web framework. Consider you might want to reuse your code with a different front-end: not a blacksheep web application, but maybe a Click CLI, for example.

[tyzhnenko]

The last piece that maked everything work is add_scoped_by_factory. Thanks!

I hope this thread will be helpful to others.

[RobertoPrevato]

I noticed a problem, though, while trying to prepare a complete example for your case. The context doesn´t expose the same service provider, so it's not possible to obtain other services created through DI inside the factory.

This is something that can be fixed easily. The idea is that the context that gets injected into factories exposes the same object that can activate services. I'll need to take a look at this.

[RobertoPrevato]

I elaborate on why you shouldn't try to inject FromHeader:

1. I recommend to instead declare a class like StorageSettings, in this example - this class is abstracted from HTTP stuff and you keep your code abstracted from your front-end logic
2. you likely want to use FromHeader in other circumstances and don´t want to resolve them only for a single scenario

Side note: as I mentioned above, there is a defect that the context doesn't expose the right provider - it should

from dataclasses import dataclass

from blacksheep import Application
from blacksheep.messages import Request
from blacksheep.server.di import dependency_injection_middleware


class BaseStorage:
    def save(self, *args):
        pass


class DefaultStorage(BaseStorage):
    ...


class SlowStorage(BaseStorage):
    ...


class FastStorage(BaseStorage):
    ...


# NOTE: Pydantic can be used, too
@dataclass
class StorageSettings:
    storage_type: str


def storage_settings_factory(context) -> StorageSettings:
    request: Request = context.scoped_services["__request__"]
    x_storage = request.get_first_header(b"X-Storage") or b"default"
    return StorageSettings(x_storage.decode())


class DataStorageFabric:
    def __init__(self, storage_settings: StorageSettings):
        x_storage = storage_settings.storage_type

        if "slow" == x_storage:
            self._storage_type = "slow"
        elif "fast" == x_storage:
            self._storage_type = "fast"
        elif "default" == x_storage:
            self._storage_type = "default"

    def storage_engine(self):
        if self._storage_type == "default":
            return DefaultStorage()
        elif self._storage_type == "slow":
            return SlowStorage()
        elif self._storage_type == "fast":
            return FastStorage()
        raise Exception("Invalid type")


class MediaProcessor:
    def __init__(self, storage_fabric: DataStorageFabric):
        self.storage: BaseStorage = storage_fabric.storage_engine()

    def make_image_better(self, image_file):
        image_object = self.storage.save(image_file)
        # some code that make the image great again :)

    def save_image(self, image_file):
        return self.storage.save(image_file)


app = Application()


def request_factory(context) -> Request:
    return context.scoped_services["__request__"]


app.middlewares.append(dependency_injection_middleware)
app.services.add_scoped(DataStorageFabric)
app.services.add_scoped_by_factory(request_factory)
app.services.add_scoped_by_factory(storage_settings_factory)
app.services.add_scoped(MediaProcessor)


@app.route("/")
def example(service: MediaProcessor):
    print(service)
    return "OK"


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, port=44555, lifespan="on")