from pymongo import MongoClient from pymongo errors import ConnectionF

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
from pymongo import MongoClient
from pymongo.errors import ConnectionFailure
from grabscrapers.settings import MONGODB_URI
class MongoDatabase:
"""Base class for mongodatabase
Implement singleton to share connection amoung whole app
How to check database availability:
https://api.mongodb.com/python/current/api/pymongo/mongo_client.html
"""
__db = None
@classmethod
def get_db(cls, dbname=None, check_availability=False, **kwargs):
if cls.__db is None:
# celery complaining about connection happened before the fork
# pass `connect=False` to MongoClient to connect on first operation
db_name = dbname or ''
cls.__db = MongoClient(MONGODB_URI.format(db_name),
connect=False, **kwargs)
# check client availability, see:
if check_availability:
try:
# The ismaster command is cheap and does not require auth.
cls.__db.admin.command('ismaster')
except ConnectionFailure as ex:
raise ex
return cls.__db
@classmethod
def close(cls):
if cls.__db:
cls.__db.close()