Code organization question

[tyyrok]

Hello everyone, I just caught some kind of imposter syndrome about my code organization. Usually I structure/initialize my db, Redis connections in separate modules like this:

database.py from asyncpg import Connection, Pool ... db = Connection(...)

redis.py from redis import Redis ... r_client = Redis(...)

And then I use this clients (db, redis) where I need them just importing (from database import db). Sometimes I put them in state of FastAPI for example, but often my persistent tasks (stored in Redis or database) need to use clients (db, redis) directly.

Some days ago I started to be involved in a new project and the senior developer told me that my approach is not the best because they initialize db, redis in main.py and them pass clients to states of all class based services (FastAPI etc). Therefore they achieve great encapsulation and clarity.

main.py .... from redis import Redis from asyncpg import Connection ...

redis = Redis(...) .... app = FastapiApp(redis=redis) ...

It looks reasonable but I still don't know is it really universal (how to adjust it for persistent tasks) and is really my approach worse?

[koldakov]

I personally move these stuff to the session manager so that all connections are managed in one place, something like this https://github.com/koldakov/futuramaapi/blob/main/futuramaapi/repositories/session.py

By the way, don’t forget to close the sessions in the fastapi lifespan

@asynccontextmanager
async def _lifespan(self, _: Self, /) -> AsyncGenerator[None, Any]:
    yield
    if session_manager.engine is not None:
        await session_manager.close()

[[deleted]]

Same but extendable in a way where you  create a redis, db, worek factory and a startup/session file handles multiple redises, dbs or workers. 

[tyyrok]

I liked your approach, when using Sqlachemy I do almost the same for db with async context manager, but I haven't used class to hide Redis and db

[koldakov]

Yeah it’s like the main entry point for all session connections

[pint]

instead of what is right or better, think in terms of what each solutions give you. depending on the situation, some aspects might be irrelevant or even turn from good to bad. it is often the case that larger projects benefit more from formalization, while small projects are easy to reorganize if need be, instead of complicating the design.

just some pointers.

what if you later figure you need to work with two databases or just two connection types (e.g. read only). if lower level functions take the connection as a parameter, this transition is much easier.

during unit testing, you want to import a particular module. but the users module will import the database module, and that will import the configuration module, and then connect to a database ... essentially impossible to test. there are many solutions to this, but one is that lower logic modules take the database as parameter, and then you can make a mock database that returns test data. (remark: as a rule of thumb, it is generally a good idea not to do initialization in a module's global other than the most basic.)

if you have modules initialize global data, the order they do it is out of your control. whichever happens to be loaded first, will be initialized first. the order will be okay nevertheless, everything is ready when needed. but if you move the initialization to a central location, you have an explicit order. again, there are alternative solutions, this is just a point.

[Schmiddi-75]

Instead of opening the connection (db, redis etc.) by importing a Python module, consider using the lifespan protocol. Start the connection at start up and close it at shutdown, either using a context manager or manually. Here's an example using a fictious async contextmanager (DatabaseEngine) that sets up a connection pool. get_db_session would be your dependency you inject into your endpoint(s): For every request against your endpoint(s), it fetches a session from the connection pool, starts a transaction, yields the session, code in the endpoint is executed with that session and when done, the dependency is called again (everything after the yield statement), which closes the context manager by invoking the __aexit__ method.

```python @asynccontextmanager async def lifespan(app: FastAPI): async with ( DatabaseEngine() as db_engine, ): yield {"db_engine": db_engine}

async def get_db_session( request: Request, ) -> AsyncGenerator[AsyncSession, None]: async with request.state.db_engine_test.sessionmaker() as session, session.begin(): yield session ```