How should I provide access to "global" objects down to other objects which are contained in a central "application" object

Question

Note This is a bit lengthy to have give a better understanding of the situation and to get some context. You might spot other architectural flaws (it's from an ancient application). I appreciate any feedback on this, but to stay on topic, I will mark the core question at the end in bold!

---

Let's assume that I have an application that uses a factory to create isolated database transactions (in my case an SQLAlchemy "Session"), but also uses other "application-global" structures like an event-bus and a metric-collector.

My current architecture feels wrong and I feel like there's a better way to do it. My architecture is a result from trying to avoiding the "Singleton" pattern because it feels to me like "hidden global state". But in a sense I want a certain "globalness". So maybe I should go for a singleton? What I don't like about the singleton is that the resulting code becomes a bit more "magical". For example, consider these two functions:

def store_in_db_no_singleton(dbconnection, item):
    dbconnection.add(item)

def store_in_db_singleton(item):
    db_connection = singleton.get_instance()
    dbconnection.add(item)

The first example makes code which is calling that function more explicit. And is makes it easier in unit-tests to inject alternative db-connection implementations.

But I digress.

For the above reason, my application now looks like the following:

class Application:
    def __init__(self, sessionmaker, eventbus, metrics):
        self.sessionmaker = sessionmaker
        self.eventbus = eventbus
        self.metrics = metrics
        self.component_1 = Component1(sessionmaker, eventbus, metrics)
        self.component_2 = Component2(sessionmaker, eventbus, metrics)

sessionmaker is responsible to generate new isolated DB transactions, Component1 and Component2 are two distinct parts of the application, each with their own problem-domain.

The way DB-transactions are handled here makes this hard to unit-test. Because the transactions are scoped to the methods inside component_1 and component_2. For example, assume component_1 has an add_user method, the implementation would look something like this:

def add_user(username):
    with self.sessionmaker() as session:
        entity = User(username)
        session.add(entity)
        session.commit()

In the unit-tests I can provide a session-maker which does a flush and rollback whenever a "commit" would be issued. This has the advantage that the SQL queries are actually sent to the test-db (which is intentional in this case) but the "commit" would never be sent.

However, because the session is scoped to the low-level add_user method, a unit test would never "see" the stored data and cannot confirm data has successfully been written. For example:

from sqlalchemy.orm.session import Session, sessionmaker
from sqlalchemy import create_engine

class RollbackSession(Session):

    def commit(self) -> None:
        super().flush()
        return super().rollback()

def test_add_user():
    engine = create_engine(test_db_dsn)
    my_session_maker = sessionmaker(bind=engine, class_=RollbackSession)
    app = Application(my_session_maker, None, None)
    app.component_1.add_user("john.doe")
    # Session has now been rolled back and it's impossible to check the DB-data

While trying to improve this I came back to the Application object (duplicated here for reference):

class Application:
    def __init__(self, sessionmaker, eventbus, metrics):
        self.sessionmaker = sessionmaker
        self.eventbus = eventbus
        self.metrics = metrics
        self.component_1 = Component1(sessionmaker, eventbus, metrics)
        self.component_2 = Component2(sessionmaker, eventbus, metrics)

I was about to change it into the following:

class Application:
    def __init__(self, sessionmaker, eventbus, metrics):
        self.sessionmaker = sessionmaker
        self.eventbus = eventbus
        self.metrics = metrics
        self.component_1 = Component1(self)
        self.component_2 = Component2(self)

But that triggered my code-smell trunk. This would mean that the two componented "reach up" to the "Application" object. And I don't know if this is acceptable practice. Maybe I am unaware of a well-known design-pattern for this?

So the core question is: Is there a cleaner way to make sessionmaker, eventbus and metrics available to the underlying components? Or is it okay like it is?

Answer 1

There are some problems with your current design.

First of all, the RollbackSession class violates the Liskov Substitution Principle because it breaks the behaviour of its parent class (the Session) class. As a consequence, your tests fail because the program does not work as expected when replacing Session objects with their subtype instances.

Then, you should put all database related operations into a separate layer in your application, which I usually call the persistence layer. I am not a Python expert. From my knowledge with other programming languages, if you want to test the persistence layer implementation with sessions, there are at least two approaches:

• Have an interface for the session then two implementations: one for the real database engine and one for in-memory database for testing
• Use a separated database for testing this layer and then reset the whole database after/before every run of the test suite.

For testing the rest of the application, there should be no interaction with the database as we will mock the persistence layer in those tests.

Last but not least, using singleton is absolutely acceptable. The importance here is to limit the places you use singletons. In many frameworks, the only place that singleton can be used is service factories, where we do the dependency injection. In your classes, it's better to have no singleton nor static method calls.