# limitations under the License.
-import logging
-from pymongo import MongoClient, errors
-from osm_common.dbbase import DbException, DbBase
-from http import HTTPStatus
-from time import time, sleep
-from copy import deepcopy
from base64 import b64decode
+from copy import deepcopy
+from http import HTTPStatus
+import logging
+from time import sleep, time
from uuid import uuid4
+from osm_common.dbbase import DbBase, DbException
+from pymongo import errors, MongoClient
+
__author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
# TODO consider use this decorator for database access retries
conn_timout = 10
def __init__(self, logger_name="db", lock=False):
- super().__init__(logger_name, lock)
+ super().__init__(logger_name=logger_name, lock=lock)
self.client = None
self.db = None
self.database_key = None
self.client = MongoClient(
config["uri"], replicaSet=config.get("replicaset", None)
)
- else:
- self.client = MongoClient(
- config["host"],
- config["port"],
- replicaSet=config.get("replicaset", None),
- )
- # TODO add as parameters also username=config.get("user"), password=config.get("password"))
# when all modules are ready
self.db = self.client[config["name"]]
if "loglevel" in config:
with self.lock:
collection = self.db[table]
db_filter = self._format_filter(q_filter)
- count = collection.count(db_filter)
+ count = collection.count_documents(db_filter)
return count
except DbException:
raise
collection = self.db[table]
if not (fail_on_empty and fail_on_more):
return collection.find_one(db_filter)
- rows = collection.find(db_filter)
- if rows.count() == 0:
+ rows = list(collection.find(db_filter))
+ if len(rows) == 0:
if fail_on_empty:
raise DbException(
"Not found any {} with filter='{}'".format(
),
HTTPStatus.NOT_FOUND,
)
+
return None
- elif rows.count() > 1:
+ elif len(rows) > 1:
if fail_on_more:
raise DbException(
"Found more than one {} with filter='{}'".format(