41 lines
1.0 KiB
Python
41 lines
1.0 KiB
Python
from typing import Any
|
|
|
|
from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase
|
|
|
|
from app.core.config import settings
|
|
|
|
|
|
class _MongoDB:
|
|
client: AsyncIOMotorClient[dict[str, Any]] | None = None
|
|
|
|
def get_database(self) -> AsyncIOMotorDatabase[dict[str, Any]]:
|
|
if self.client is None:
|
|
raise RuntimeError("MongoDB client has not been initialized")
|
|
return self.client[settings.MONGODB_DB_NAME]
|
|
|
|
|
|
mongodb = _MongoDB()
|
|
|
|
|
|
async def init_mongodb() -> None:
|
|
mongodb.client = AsyncIOMotorClient(settings.MONGODB_URI)
|
|
await ensure_mongodb_available()
|
|
|
|
|
|
async def close_mongodb() -> None:
|
|
client = mongodb.client
|
|
mongodb.client = None
|
|
if client is not None:
|
|
client.close()
|
|
|
|
|
|
async def get_database() -> AsyncIOMotorDatabase[dict[str, Any]]:
|
|
return mongodb.get_database()
|
|
|
|
|
|
async def ensure_mongodb_available() -> None:
|
|
client = mongodb.client
|
|
if client is None:
|
|
raise RuntimeError("MongoDB client has not been initialized")
|
|
await client.admin.command("ping")
|