Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def build_connections(self, name):
connections = []
configs = self.db_configs(name)
for db in configs:
if db.scheme not in self.SUPPORTED_SCHEMES:
raise UnsupportedDriver('%s is not a supported driver' % db.scheme)
elif db.scheme == 'postgres' or db.scheme == 'redshift':
import jardin.database.drivers.pg as driver
elif db.scheme == 'mysql':
import jardin.database.drivers.mysql as driver
elif db.scheme == 'sqlite':
import jardin.database.drivers.sqlite as driver
elif db.scheme == 'snowflake':
import jardin.database.drivers.sf as driver
connections.append(driver.DatabaseConnection(db, name))
return connections
def execute(self, *query):
return super(DatabaseConnection, self).execute(*query)