I'm trying to build dynamic connections to databases. This because I just now and then have to query them to get some data. The sources change over time, so I'm not in favor of adding them to settings.

Currently I have the following code:

    DBTYPE = (
        ('postgres', ('PostgreSQL')),
        ('mysql', ('MySQL')),
        ('mysql', ('MariaDB')),
        ('oracle', ('Oracle')),
        ('mssql', ('MSSQL')),
        ('sqlite', ('SQLite')),
    )
    URLTEMPLATES = (
        ('postgres', ('postgres://{USER}:{PASSWORD}@{HOST}:{PORT}/{NAME}')),
        ('mysql', ('mysql://{USER}:{PASSWORD}@{HOST}:{PORT}/{NAME}')),
        ('oracle', ('oracle://{USER}:{PASSWORD}@{HOST}:{PORT}/{NAME}')),
        ('mssql', ('mssql://{USER}:{PASSWORD}@{HOST}:{PORT}/{NAME}')),
        ('sqlite', ('sqlite:///{HOST}')),
    )
    dburl = [item for item in URLTEMPLATES if item[0] == self.engine]
    self.db_url = dburl.format(**template_values)
    dbsetting = dj_database_url.parse(self.db_url,conn_max_age=600,conn_health_checks=True,)
    


     database_id = self.name 
                newDatabase = {}
                for conn in connections.all():
                    print(conn)
                try:
                    newDatabase["id"] = database_id
                    newDatabase['ENGINE'] = dbsetting['ENGINE']
                    newDatabase['NAME'] = dbsetting['NAME']
                    newDatabase['USER'] = dbsetting['USER']
                    newDatabase['PASSWORD'] = dbsetting['PASSWORD']
                    newDatabase['HOST'] = dbsetting['HOST']
                    newDatabase['PORT'] = dbsetting['PORT']
                    newDatabase['ATOMIC_REQUESTS'] = True
                    newDatabase['TIME_ZONE'] = 'Europe/Paris'
                    newDatabase['CONN_HEALTH_CHECKS'] = False
                    newDatabase['CONN_MAX_AGE'] = 600
                    newDatabase['OPTIONS'] = {}
                    newDatabase['AUTOCOMMIT'] = False
                    connections.databases[database_id] = newDatabase
    
                   cursor = connections[database_id].cursor()
                   cursor.close()

What I would like to achieve is, that when Django is unable to connect, it just responds with a message that connection could not be established.

Now I get, for example:

Exception Type: OperationalError
Exception Value:    
(1045, "Access denied for user 'test'@'localhost' (using password: YES)")

In those case I would just like to respond with an error that the connection could not be established.

How could I go about this?

0

There are 0 best solutions below