Support for Registry lazy value factories

[hynek]

This idea is based on #8 (comment) and @elfjes concerns about container lifetimes.

Currently, svcs allows to register on_registry_close cleanups, but that's only necessary if an object created outside of svcs. For me that's common, but maybe just because I'm used to it.

The idea is to add a third kind of service, the singleton. A singleton is a factory that is only called once it is needed and then it lives as long as the registry.

So instead of:

def create_engine_very_complex(base: Service):
    engine = MyEngine(base)
    # some more logic
    ...
    return engine

engine = create_engine_very_complex(some_service)

registry.register_value(Engine, engine, on_registry_close=engine.dispose)

You'd write:

def create_engine_very_complex(container: svcs.Container):
    engine = MyEngine(container.get(Service))
    # some more logic
    ...
    yield engine

    engine.dispose()

registry.register_singleton(Engine, create_engine_very_complex)

This allows to late-bind base/Service and integrates the engine more seamlessly into the whole life-cycle

Is this fit your vision @elfjes?

---

I'm also open to a better name, because surely there's some hairsplitting to be done about the term "singleton".

[elfjes]

Yeah that looks good 😊

As for the name, could be a longlived_service, or maybe an application_service (although that might be confusing)?

[hynek]

better name idea: register_lazy_value_factory courtesy of @flaper87

[madjar]

So, as a datapoint, here is what I ended up adding in my fastapi lifespan function, as a way to avoid initializing an engine as a side-effect of initializing the registry.

@functools.cache
def get_engine() -> AsyncEngine:
    return create_async_engine("...")

registry.register_factory(
    AsyncEngine,
    get_engine,
    on_registry_close=lambda: get_engine().dispose,
)

This allows me to easily replace the engine in the tests with

lifespan.registry.register_value(AsyncEngine, pytest_managed_postgres_engine)

[mtvx]

@madjar What did you end up with here? How did you resolve getting reference on the instance on registry close?

As a solution, if the on_registry_close passed the service instance (ie. AsyncEngine there) to the callable (like ping does), we could just instance.close() it. @hynek what do you think?

Another thing is that if the returned value from factory was context manager, it would get closed when the first container closes. To avoid that, one would need to pass enter=False. Fair enough, there's a solution, but cleaner interface (e.g. that register_singleton() or alike) could be nice.

[madjar]

Here is what I ended up with. In short:

• A singleton decorator so that multiple call to the factory return the same thing.
• A special case in that singleton decorator to handle "give it to me if it exists but don't create it" to handle on_registry_close. (Specifically, if the factory function gets passed None instead of a svcs.Container, then won't try to create a new value)
• The dispose_engine engine function handles that special case. (Specifically, ignore None).

It's not pretty, and I don't love it, but it did the job well enough to move on. The main issue is that I have to handle the lifecycle of that object myself (has it been created, does it need to be deleted), when that's something that feels like the job of svcs. Now that I think about it, it's as if we want two different level of lifecycles (per-request and per-app) and maybe having two Registries, one deferring to the other, would help here.

Here is more or less what the code looks like:

def singleton(f: Callable[[svcs.Container], T]) -> Callable[[svcs.Container | None], T]:
    value = None

    def wrapper(svcs_container: svcs.Container | None):  # type: ignore[no-untyped-def]
        nonlocal value

        if not value and svcs_container:
            value = f(svcs_container)
        return value

    return cast(Callable[[svcs.Container | None], T], wrapper)

  @singleton
  def get_engine(svcs_container: svcs.Container) -> AsyncEngine:
      ...
      return engine

  async def dispose_engine() -> None:
      engine = get_engine(None)
      if engine:
          await engine.dispose()

  registry.register_factory(
      AsyncEngine,
      get_engine,
      on_registry_close=dispose_engine,
  )

[mtvx]

Thanks for the quick reply. It's certainly not too pretty 😅 But good that it works for you, and you have implemented it as a re-usable piece of code (decorator) so thanks for sharing!

two different level of lifecycles (per-request and per-app)

Basically it is there already, as registry (per-app) and container (per-request) level. The only thing that's missing is getting the instance for registry level closing (either via your hack, or if the instance was passed to the on_registry_close callback).

Of course, it could be of dynamic depth, ie. you could create deeply nested containers... don't know if anyone would have that need.

maybe having two Registries, one deferring to the other, would help here.

Not sure how that would play out. I get this feeling it could get complicated(?)

[mtvx]

Passing stuff to the on close callable should be easy. The callable is not bound to the value returned from the factory, though. But, it should be easy to pass registry instance as oc(self) and then registry.get_registered_service_for() to find the instance?

[mtvx]

Some thought about nested containers seems to already exist in #70.

What's suggested there would solve also the troubles highlighted in this thread.