41 lines
1.1 KiB
Python
41 lines
1.1 KiB
Python
from motor.motor_asyncio import AsyncIOMotorClient
|
|
from redis import asyncio as aioredis
|
|
|
|
# 数据库配置
|
|
MONGODB_URL = "mongodb://lab:[email protected]:27017/lab"
|
|
REDIS_URL = "redis://:Obscura@[email protected]:6379"
|
|
|
|
# 数据库连接类
|
|
class Database:
|
|
"""数据库连接管理类"""
|
|
client: AsyncIOMotorClient = None
|
|
|
|
db = Database()
|
|
redis_client = None
|
|
|
|
async def get_database():
|
|
"""获取数据库连接"""
|
|
if db.client is None:
|
|
await connect_to_mongo()
|
|
return db.client["lab"]
|
|
|
|
async def connect_to_mongo():
|
|
try:
|
|
db.client = AsyncIOMotorClient(MONGODB_URL)
|
|
# 验证连接
|
|
await db.client.admin.command('ping')
|
|
print("Successfully connected to MongoDB")
|
|
except Exception as e:
|
|
print(f"Could not connect to MongoDB: {e}")
|
|
raise
|
|
|
|
async def close_mongo_connection():
|
|
if db.client:
|
|
db.client.close()
|
|
|
|
async def init_redis():
|
|
global redis_client
|
|
if redis_client is None:
|
|
redis_client = await aioredis.from_url(REDIS_URL, db=200)
|
|
return redis_client
|