MongoConnection.py

Go to the documentation of this file.

import os
import ssl
from pymongo import MongoClient
from dotenv import load_dotenv, dotenv_values
 
load_dotenv()
 
class mongo_connection:
    """
    Makes mongo connection and depending of case, returns the right collection
    """
    def __init__(self):
        pass
 
    def mongo_conn_sales(self):
        """
        Connect with sales collection
 
        Args:
            None
        Returns collection (str)
        """
        client = self.mongo_conn()
        db_name = os.environ.get("DB")
        collection_name = os.environ.get("COLLECTION")
        if client:
            db = client[db_name]
            collection = db[collection_name] #collection to upload sales
            return collection
        raise Exception("Mongo Connection Fails.")
 
    def mongo_conn_catalogue(self):
        """
        Connect with catalogue collection
 
        Args:
            None
        Returns catalogue_collection (str)
        """
        client = self.mongo_conn()
        db_name = os.environ.get("DB")
        catalog_name = os.environ.get("AS_COLLECTION")
        if client:
            db = client[db_name]
            catalogue_collection = db[catalog_name]
            return catalogue_collection
        raise Exception("Mongo Connection Fails.")
    
    def mongo_conn_catalogue_gen(self):
        """
        Connect with catalogue collection
 
        Args:
            None
        Returns catalogue_collection (str)
        """
        client = self.mongo_conn()
        db_name = os.environ.get("DB")
        catalog_name = os.environ.get("CAT_COLLECTION")
        if client:
            db = client[db_name]
            catalogue_collection = db[catalog_name]
            return catalogue_collection
        raise Exception("Mongo Connection Fails.")
 
    def mongo_conn_snapshots(self):
        """
        Connect with snapshots collection
 
        Args:
            None
        Returns collection (str)
        """
        client = self.mongo_conn()
        db_name = os.environ.get("SNAPSHOTS_DB")
        collection_name = os.environ.get("SNAPSHOTS")
        if client:
            db = client[db_name]
            collection = db[collection_name]
            return collection
        raise Exception("Mongo Connection Fails.")
 
    def mongo_conn_formats(self):
        """
        Connect with formats collection
 
        Args:
            None
        Returns collection (str)
        """
        client = self.mongo_conn()
        db_name = os.environ.get("SNAPSHOTS_DB")
        formats_collection = os.environ.get("FORMATS")
        if client:
            db = client[db_name]
            collection = db[formats_collection]
            return collection
        raise Exception("Mongo Connection Fails.") 
 
    def mongo_conn_ftp(self):
        """
        Connect with ftp collection
 
        Args:
            None
        Returns collection (str)
        """
        client = self.mongo_conn()
        db_name = os.environ.get("DB")
        formats_collection = os.environ.get("FTP_ACCESS")
        if client:
            db = client[db_name]
            collection = db[formats_collection]
            return collection
        raise Exception("Mongo Connection Fails.")
 
    def mongo_conn(self):
        """
        Connect with the main MongoDB database
 
        Args:
            None
        Returns client (mongo client)
        """
        uri = os.environ.get("MONGO_GLIDER")
        if "=true&w=majority" not in uri:
            uri+= "=true&w=majority"
        # print("Este es el URI", uri)
        try:
            client = MongoClient(uri,ssl_cert_reqs=ssl.CERT_NONE)
        except:
            client = MongoClient(uri)
        if client:
            return client
        raise Exception("Mongo Connection Fails.")
 
# def connection_per_module()