Connecting MongoDB using Motor in FastAPI

Question

What are the best practices for connecting to a database in FastAPI?

To provide some context, I want to write code to connect to a MongoDB database using Motor. My idea is to create a single connection and use it in all the controllers that need it through Dependency Injection, but I am not quite sure how to do it. So let me show you a simple code example to illustrate this idea in a nutshell:

database.py:

from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase
from dotenv import dotenv_values
class Database:
 _client: AsyncIOMotorClient | None = None
 _db: AsyncIOMotorDatabase | None = None
 @staticmethod
 def connect() -> None:
 config = dotenv_values(".env")
 Database._client = AsyncIOMotorClient(config["ATLAS_URI"])
 Database._db = Database._client[config["DB_NAME"]]
 @staticmethod
 def close() -> None:
 if Database._client is not None:
 Database._client.close()
 else:
 raise ConnectionError("Client not connected")
 @staticmethod
 def get_db() -> AsyncIOMotorDatabase:
 if Database._db is not None:
 return Database._db
 else:
 raise ConnectionError("Database not connected")

main.py:

from fastapi import FastAPI, Depends
import uvicorn
from database import Database
from contextlib import asynccontextmanager
from motor.motor_asyncio import AsyncIOMotorDatabase
@asynccontextmanager
async def lifespan(app: FastAPI):
 Database.connect()
 yield
 Database.close()
app = FastAPI(lifespan=lifespan)
@app.get("/")
async def main(db: AsyncIOMotorDatabase = Depends(Database.get_db)):
 await db["books"].insert_one({"hello": "world"})
 return "Done"
if __name__ == "__main__":
 uvicorn.run("main:app", reload=True)

As you can see, i have a Database class that is designed to manage the database connection. In main.py, within the lifespan function, we start the connection to the MongoDB database before the app starts running and close it when the app stops. Finally, as an example, we have a small endpoint that obtains the database instance through Dependency Injection and creates a simple document in a collection called 'books'.

The idea is to divide the code in the future into Models, Controllers, and Services to create better code. However, this isn't the focus of the current question, so I've chosen not to provide an example code.

I would like to know what you think about my solution. Are there any ways to improve it? Am I following the best practices? Can you identify any potential issues? Any suggestions are welcome. If you have another approach, feel free to share it.

Answer 1

@classmethod

In connect() and close(), it's weird to see you assigning these:

• Database._client
• Database._db

Prefer the idiomatic cls._client and cls._db.

We use @staticmethod for something with no deps, such as def sqrt(x: float), and @classmethod for something which interacts with the class, such as these two. And we don't only do that due to inheritance concerns.

close both

 Database._client.close()

I guess the ._db doesn't offer a close() method? Ok. I'm just surprised, that's fine. It's not a library I'm familiar with.

Don't we want to assign a None on top of the now closed _client?

I wonder if an unconditional call to .close() would suffice, and if caller messed up an error will be raised without doing a special check here.

context handler

This whole Public API seems a little on the clunky side. Wouldn't caller prefer to invoke using with? And then we get auto-close.

OIC, later on lifespan() roughly does that. Maybe put lifespan() in the database.py module, with a private database class which only lifespan() needs to worry about? Maybe don't name the function "lifespan"? Given that it is the public access point to all this.

Answer 2

I am quite unfamiliar with Mongodb and best practices so I won't delve too much in the specifics. To enhance what @J_H has written:

A close method is typically run as part of cleanup code. I believe it is okay to raise a simple warning here rather than an exception. Thus, logging.warning sounds sufficient to me. You do not intend to have multiple connection handlers open at the same time, in which case it would make sense to draw attention to logic errors (attempting to close a connection that isn't open in the first place).

Sadly I am not familiar with that lib. But in fact, I suspect that an InvalidOperation exception would be raised by the lib anyway. Check the doc to be sure, and test. Therefore I believe your code is superfluous:

@staticmethod
def close() -> None:
 if Database._client is not None:
 Database._client.close()
 else:
 raise ConnectionError("Client not connected")

Context manager

I guess the lib can be used with a context manager to close the connection automatically to upon exit. But when dealing with libs that don't have that facility, you can still implement your own using contextlib.

Here is an example borrowed from a Digital Ocean tutorial (source) that shows how to implement the feature for SQLite:

from contextlib import closing
with closing(sqlite3.connect("aquarium.db")) as connection:
 with closing(connection.cursor()) as cursor:
 rows = cursor.execute("SELECT 1").fetchall()
 print(rows)

Naming conventions

"Database" is much too broad as name. It's not like you are coding a whole ORM. This is just a wrapper and it is MongoDB-specific, so the name should reflect that in being less generic and more descriptive.

Misc

I have no idea about locking and concurrency here. Transactions? Sorry, I am more familiar with relational databases. But how can you accommodate the need?

Improvements and reuse

Connection pooling can be even more flexible. I believe you can switch to another database while sharing the same connection pool. You may not have a use case for that right now, but if you intend to distribute your code and make it more reusable, this could be a TODO item in your list.

However this code is not really portable, since it is tied to the presence of a .env file by design. You are not checking the existence of the .env file. What would happen if the file were missing? Any warning, crash, or fallback values?

Conclusion

After pondering all this, I am still trying to understand the added value of your wrapper. When I look at the Motor doc, your code does not seem to reduce complexity or decrease the amount of code needed. In fact your class does very little... as it stands.

I believe there is nothing that prevents you from referring to a single connection without this class, using the regular Motor functions. To reuse an object across your app, you either instantiate it at a central place or put it in a separate file, that you import as needed. Multiple import is no problem as the dependency will be loaded only once, when called for the first time.

Imagine you are doing a job interview for a dev role, what would be the compelling reason for adopting your code vs sticking to vanilla code?