【自分用メモ】MongoDBAtlasのDataAPIラッパーをpythonで実装

MongoDBAtlasのDataAPIラッパーをpythonで実装

APIの方が疎結合的な意味で便利かなとか思ってましたが、どうやら無料プランだと制限があるようなので、pymongoに戻します
一応メモしておきます

実装

URLとAPIKEYは変えてくださいネ

import json
import requests
import copy


class MongoRepository:
    URL = "https://XXXXX.data.mongodb-api.com/app/XXXXX/endpoint/data/v1/action/"

    PAYLOAD = {
        "collection": "",
        "database": "datastore",
        "dataSource": "Cluster0",
    }
    HEADERS = {
        'apiKey': 'XXXXX',
        'Access-Control-Request-Headers': '*',
        'Content-Type': 'application/json'
    }

    def _select_common(self, method, repository_key, filter={}):
        payload = copy.deepcopy(self.PAYLOAD)
        payload["collection"] = repository_key
        payload["filter"] = filter
        str_data = json.dumps(payload, ensure_ascii=False)
        return requests.post(self.URL+method, data=str_data.encode('utf-8'), headers=self.HEADERS).json()

    def select(self, repository_key, filter={}):
        return self._select_common("findOne", repository_key, filter).get("document")

    def select_all(self, repository_key, filter={}):
        return self._select_common("find", repository_key, filter).get("documents")

    def insert(self, repository_key, data):
        payload = copy.deepcopy(self.PAYLOAD)
        payload["collection"] = repository_key
        payload["document"] = data
        str_data = json.dumps(payload, ensure_ascii=False)
        return requests.post(self.URL+"insertOne", data=str_data.encode('utf-8'), headers=self.HEADERS).json()

    def insert_all(self, repository_key, data):
        payload = copy.deepcopy(self.PAYLOAD)
        payload["collection"] = repository_key
        payload["documents"] = data
        str_data = json.dumps(payload, ensure_ascii=False)
        return requests.post(self.URL+"insertMany", data=str_data.encode('utf-8'), headers=self.HEADERS).json()

    def _update_common(self, method, repository_key, filter, data):
        payload = copy.deepcopy(self.PAYLOAD)
        payload["collection"] = repository_key
        payload["filter"] = filter
        payload["update"] = {}
        payload["update"]["$set"] = data
        str_data = json.dumps(payload, ensure_ascii=False)
        return requests.post(self.URL+method, data=str_data.encode('utf-8'), headers=self.HEADERS).json()

    def update(self, repository_key, filter, data):
        return self._update_common("updateOne", repository_key, filter, data)

    def update_all(self, repository_key, filter, data):
        return self._update_common("updateMany", repository_key, filter, data)

    def _delete_common(self, method, repository_key, filter):
        payload = copy.deepcopy(self.PAYLOAD)
        payload["collection"] = repository_key
        payload["filter"] = filter
        str_data = json.dumps(payload, ensure_ascii=False)
        return requests.post(self.URL+method, data=str_data.encode('utf-8'), headers=self.HEADERS).json()

    def delete(self, repository_key, filter):
        return self._delete_common("deleteOne", repository_key, filter)

    def delete_all(self, repository_key, filter):
        return self._delete_common("deleteMany", repository_key, filter)

使い方

mdb = MongoRepository()
print(mdb.insert("test", {"aa": 123}))
print(mdb.delete_all("test",{}))
print(mdb.select("test"))
print(mdb.select_all("test", {"aa": 123}))
print(mdb.update("test", {"aa": 123}, {"bb": "gdfgarg"}))
print(mdb.delete("test", {"bb": "gdfgarg"}))