feat: 实现MongoDB URI方式连接,并统一数据库连接代码。

This commit is contained in:
春河晴
2025-03-10 14:48:43 +09:00
parent c9f12446c0
commit 4baa6c6f0a
9 changed files with 82 additions and 121 deletions

View File

@@ -6,20 +6,44 @@ from pymongo import MongoClient
class Database:
_instance: Optional["Database"] = None
def __init__(self, host: str, port: int, db_name: str, username: Optional[str] = None, password: Optional[str] = None, auth_source: Optional[str] = None):
if username and password:
def __init__(
self,
host: str,
port: int,
db_name: str,
username: Optional[str] = None,
password: Optional[str] = None,
auth_source: Optional[str] = None,
uri: Optional[str] = None,
):
if uri and uri.startswith("mongodb://"):
# 优先使用URI连接
self.client = MongoClient(uri)
elif username and password:
# 如果有用户名和密码,使用认证连接
# TODO: 复杂情况直接支持URI吧
self.client = MongoClient(host, port, username=username, password=password, authSource=auth_source)
self.client = MongoClient(
host, port, username=username, password=password, authSource=auth_source
)
else:
# 否则使用无认证连接
self.client = MongoClient(host, port)
self.db = self.client[db_name]
@classmethod
def initialize(cls, host: str, port: int, db_name: str, username: Optional[str] = None, password: Optional[str] = None, auth_source: Optional[str] = None) -> "Database":
def initialize(
cls,
host: str,
port: int,
db_name: str,
username: Optional[str] = None,
password: Optional[str] = None,
auth_source: Optional[str] = None,
uri: Optional[str] = None,
) -> "Database":
if cls._instance is None:
cls._instance = cls(host, port, db_name, username, password, auth_source)
cls._instance = cls(
host, port, db_name, username, password, auth_source, uri
)
return cls._instance
@classmethod