创建一个 python 类来返回数据库连接

如何解决创建一个 python 类来返回数据库连接

背景

经过一番努力，我设法为 Amazon DocumentDb 创建了一个集群。现在我想编写一个简单的 python 类，当实例化时返回一个客户端连接并允许我插入一个文档。插入文档完成后，它会安全地关闭连接。

经过更多的努力，我设法使以下内容起作用。

我的代码

import pymongo
import sys

client = pymongo.MongoClient(
    "mongodb://foousername:[e-mail]:27017/?ssl=true&ssl_ca_certs=rds-combined-ca-bundle.pem&replicaSet=rs0&readPreference=secondaryPreferred&retryWrites=false"
)

##Specify the database to be used
db = client.sample_database

##Specify the collection to be used
col = db.sample_collection


col.insert_one({"id123": "Amazon DocumentDB"})


##Close the connection
client.close()

问题

这段代码很简单，它创建了一个连接，然后我可以指定要插入文档的数据库和集合。插入部分对我来说很有意义。我正在努力将整个客户端字符串封装到一个方法中。我尝试了以下方法。

def get_mongodb_connection(
    host: str,port: Union[str,int],user: str,password: str,database: str,):
    return pymongo.MongoClient(
        host=host,port=int(port),username=user,password=password,)[database]

x = get_mongodb_connection("docdb-poc-test.cluster-z7zzzzzzzztv.us-east-1.docdb.amazonaws.com","27017","foousername","foopassword","sample_database")  

以下 x.list_collection_names() 导致 ServerSelectionTimeoutError

我知道我的原始代码有效并且能够让我连接并让我在数据库中插入文档，但我想创建一个 python 类，该类将有一个返回连接的方法，因为它可以让我轻松地做更多的事情.

我非常感谢这里的任何意见。

解决方法

没有看到你的其余代码，只尽可能接近你的代码，我为你想出了这个：

import pymongo
import sys

class MyMongoDBClass:
    def __init__(
        self,host: str,port: Union[str,int],user: str,password: str,database: str
        ):

        self.host = host
        self.port = port
        self.user = user
        self.password = password
        self.database = database


        self._client = pymongo.MongoClient(
            f'{self.user}{self.password}@{self.host}:{self.port}/?ssl=true&ssl_ca_certs=rds-combined-ca-bundle.pem&replicaSet=rs0&readPreference=secondaryPreferred&retryWrites=false'
        )
        self._db = self._client[self.database]
        self._collection = self._db.sample_collection

    def close():
        _client.close()

    def insert_one(self,key: str,value: str):
        self._collection.insert_one({key: value})

if __name__ == '__main__':
    my_db = MyMongoDBClass(
        'docdb-poc-test.cluster-z7zzzzzzzztv.us-east-1.docdb.amazonaws.com','27017','foousername','foopassword','sample_database'
    )

    my_db.insert_one('id123','Amazon DocumentDB')

    my_db.close()