Skip to content

mongodb_data_insert

MongoDB data insert

DB

A class to interact with a MongoDB database.

Source code in redesign_pipeline/db_utils/mongodb_data_insert.py 
 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
class DB:
    """A class to interact with a MongoDB database."""

    def __init__(self,db,collection) -> None:
        """
        Initialize the DB class.

        Args:
            db (str): The name of the MongoDB database.
            collection (str): The name of the collection within the database.
        """
        mongo_url="mongodb://localhost:27017"
        mongo_db= MongoClient(mongo_url)
        self.db = mongo_db[db]
        self.coll=self.db[collection]

    def put_to_db(self, data):
        """
        Insert data into the MongoDB collection.

        Args:
            data (dict): The data to be inserted into the collection.
        """
        ret_id = self.coll.insert_one(data)

    def flush_db(self):
        """Delete all documents from the MongoDB collection."""
        ret = self.coll.delete_many({})
        logger.info(f"{ret.deleted_count}, Documents deleted!")

    def query_data(self,job_id):
        query = {'job_id':job_id}
        print(query)
        estimates = self.coll.find(query)
        estimates = pd.DataFrame(estimates)
        return estimates

__init__(db, collection)

Initialize the DB class.

Parameters:

Name Type Description Default
db str

The name of the MongoDB database.

 required
collection str

The name of the collection within the database.

 required
Source code in redesign_pipeline/db_utils/mongodb_data_insert.py 
 9
10
11
12
13
14
15
16
17
18
19
20
def __init__(self,db,collection) -> None:
    """
    Initialize the DB class.

    Args:
        db (str): The name of the MongoDB database.
        collection (str): The name of the collection within the database.
    """
    mongo_url="mongodb://localhost:27017"
    mongo_db= MongoClient(mongo_url)
    self.db = mongo_db[db]
    self.coll=self.db[collection]

flush_db()

Delete all documents from the MongoDB collection.

Source code in redesign_pipeline/db_utils/mongodb_data_insert.py 
31
32
33
34
def flush_db(self):
    """Delete all documents from the MongoDB collection."""
    ret = self.coll.delete_many({})
    logger.info(f"{ret.deleted_count}, Documents deleted!")

put_to_db(data)

Insert data into the MongoDB collection.

Parameters:

Name Type Description Default
data dict

The data to be inserted into the collection.

 required
Source code in redesign_pipeline/db_utils/mongodb_data_insert.py 
22
23
24
25
26
27
28
29
def put_to_db(self, data):
    """
    Insert data into the MongoDB collection.

    Args:
        data (dict): The data to be inserted into the collection.
    """
    ret_id = self.coll.insert_one(data)