Singleton provider

Singleton provider provides single object. It memorizes the first created object and returns it on the rest of the calls.

from dependency_injector import containers, providers


class UserService:
    ...


class Container(containers.DeclarativeContainer):

    user_service_provider = providers.Singleton(UserService)


if __name__ == "__main__":
    container = Container()

    user_service1 = container.user_service_provider()
    user_service2 = container.user_service_provider()
    assert user_service1 is user_service2

Singleton provider handles dependencies injection the same way like a Factory provider.

Note

Singleton provider makes dependencies injection only when creates an object. When an object is created and memorized Singleton provider just returns it without applying injections.

Specialization of the provided type and abstract singletons work the same like like for the factories:

Singleton provider scope is tied to the container. Two different containers will provider two different singleton objects:

from dependency_injector import containers, providers


class UserService:
    ...


class Container(containers.DeclarativeContainer):

    user_service_provider = providers.Singleton(UserService)


if __name__ == "__main__":
    container1 = Container()
    user_service1 = container1.user_service_provider()
    assert user_service1 is container1.user_service_provider()

    container2 = Container()
    user_service2 = container2.user_service_provider()
    assert user_service2 is container2.user_service_provider()

    assert user_service1 is not user_service2

Resetting memorized object

To reset a memorized object you need to call the .reset() method of the Singleton provider.

from dependency_injector import containers, providers


class UserService:
    ...


class Container(containers.DeclarativeContainer):

    user_service = providers.Singleton(UserService)


if __name__ == "__main__":
    container = Container()

    user_service1 = container.user_service()

    container.user_service.reset()

    user_service2 = container.user_service()
    assert user_service2 is not user_service1

Note

Resetting of the memorized object clears the reference to it. Further object’s lifecycle is managed by the garbage collector.

You can use .reset() method with a context manager. Memorized instance will be reset on both entering and exiting a context.

from dependency_injector import containers, providers


class UserService:
    ...


class Container(containers.DeclarativeContainer):

    user_service = providers.Singleton(UserService)


if __name__ == "__main__":
    container = Container()

    user_service1 = container.user_service()

    with container.user_service.reset():
        user_service2 = container.user_service()

    user_service3 = container.user_service()

    assert user_service1 is not user_service2
    assert user_service2 is not user_service3
    assert user_service3 is not user_service1

Context manager .reset() returns resetting singleton provider. You can use it for aliasing.

with container.user_service.reset() as user_service:
    ...

Method .reset() resets only current provider. To reset all dependent singleton providers call .full_reset() method.

from dependency_injector import containers, providers


class Database:
    ...


class UserService:
    def __init__(self, db: Database):
        self.db = db


class Container(containers.DeclarativeContainer):

    database = providers.Singleton(Database)

    user_service = providers.Singleton(UserService, db=database)


if __name__ == "__main__":
    container = Container()

    user_service1 = container.user_service()

    container.user_service.full_reset()

    user_service2 = container.user_service()
    assert user_service2 is not user_service1
    assert user_service2.db is not user_service1.db

Method .full_reset() supports context manager interface like .reset() does.

with container.user_service.full_reset() as user_service:
    ...

See also: Reset container singletons.

Using singleton with multiple threads

Singleton provider is NOT thread-safe. You need to explicitly establish a synchronization for using the Singleton provider in the multi-threading application. Otherwise you could trap into the race condition problem: Singleton will create multiple objects.

There are two thread-safe singleton implementations out of the box:

  • ThreadSafeSingleton - is a thread-safe version of a Singleton provider. You can use in multi-threading applications without additional synchronization.

  • ThreadLocalSingleton - is a singleton provider that uses thread-locals as a storage. This type of singleton will manage multiple objects - the one object for the one thread.

import threading
import queue

from dependency_injector import containers, providers


def put_in_queue(example_object, queue_object):
    queue_object.put(example_object)


class Container(containers.DeclarativeContainer):

    thread_local_object = providers.ThreadLocalSingleton(object)

    queue_provider = providers.ThreadSafeSingleton(queue.Queue)

    put_in_queue = providers.Callable(
        put_in_queue,
        example_object=thread_local_object,
        queue_object=queue_provider,
    )

    thread_factory = providers.Factory(
        threading.Thread,
        target=put_in_queue.provider,
    )


if __name__ == "__main__":
    container = Container()

    n = 10
    threads = []
    for thread_number in range(n):
        threads.append(
            container.thread_factory(name="Thread{0}".format(thread_number)),
        )
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()

    all_objects = set()
    while not container.queue_provider().empty():
        all_objects.add(container.queue_provider().get())

    assert len(all_objects) == len(threads) == n
    # Queue contains same number of objects as number of threads where
    # thread-local singleton provider was used.

Implementing scopes

To implement a scoped singleton provider use a Singleton provider and reset its scope when needed.

from dependency_injector import containers, providers
from flask import Flask, current_app


class Service:
    ...


class Container(containers.DeclarativeContainer):

    service_provider = providers.ThreadLocalSingleton(Service)


def index_view():
    service_1 = current_app.container.service_provider()
    service_2 = current_app.container.service_provider()
    assert service_1 is service_2
    print(service_1)
    return "Hello  World!"


def teardown_context(request):
    current_app.container.service_provider.reset()
    return request


container = Container()

app = Flask(__name__)
app.container = container
app.add_url_rule("/", "index", view_func=index_view)
app.after_request(teardown_context)


if __name__ == "__main__":
    app.run()

The output should look like this (each request a Service object has a different address):

 * Serving Flask app "singleton_scoped" (lazy loading)
 * Environment: production
   WARNING: This is a development server. Do not use it in a production deployment.
   Use a production WSGI server instead.
 * Debug mode: off
 * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
<__main__.Service object at 0x1099a9d90>
127.0.0.1 - - [25/Aug/2020 17:33:11] "GET / HTTP/1.1" 200 -
<__main__.Service object at 0x1099a9cd0>
127.0.0.1 - - [25/Aug/2020 17:33:17] "GET / HTTP/1.1" 200 -
<__main__.Service object at 0x1099a9d00>
127.0.0.1 - - [25/Aug/2020 17:33:18] "GET / HTTP/1.1" 200 -
<__main__.Service object at 0x1099a9e50>
127.0.0.1 - - [25/Aug/2020 17:33:18] "GET / HTTP/1.1" 200 -
<__main__.Service object at 0x1099a9d90>
127.0.0.1 - - [25/Aug/2020 17:33:18] "GET / HTTP/1.1" 200 -