Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def db_connect(args):
if args.type == "mysql" or args.type == "mariadb":
import mysql.connector
try:
connection = mysql.connector.connect(
user=args.user,
password=args.password,
database=args.db)
except mysql.connector.Error as err:
print(colorize("red", "[ERROR] {}".format(err)))
return None
elif args.type == "mssql":
import pymssql
try:
connection = pymssql.connect(server="localhost", database=args.db)
except pymssql.Error as err:
print(colorize("red", "[ERROR] {}".format(err)))
return None
elif args.type == "pgsql":
import psycopg2
try:
connection = psycopg2.connect(
"dbname='{}' user='{}' password='{}'".format(
args.db, args.user, args.password))
except psycopg2.Error as err:
print(colorize("red", "[ERROR] {}".format(err)))
return None
elif args.type == "oracle":
import cx_Oracle
try:
connection = cx_Oracle.connect(
def connect_wind(self):
import pymssql
connection = pymssql.connect('192.168.1.181', 'sa', 'Nm,.hjkl', 'wind')
cursor = connection.cursor()
self.logger.debug('Connected to MSSQL Database wind on 192.168.1.181')
self.__dict__.update({'connection': connection, 'cursor': cursor})
def make_connection(self):
self.__connection = pymssql.connect(server=self.__config.get('AzureDB', 'server_URI'), user=self.__config.get('AzureDB', 'username'), password=self.__config.get('AzureDB', 'password'), database=self.__config.get('AzureDB', 'database_1'))
def _connect(self, query):
'''
Connect to MDT and dump contents of dbo.ComputerIdentity database
'''
if not self.conn:
self.conn = pymssql.connect(server=self.mdt_server + "\\" + self.mdt_instance, user=self.mdt_user, password=self.mdt_password,
database=self.mdt_database)
cursor = self.conn.cursor()
cursor.execute(query)
self.mdt_dump = cursor.fetchall()
self.conn.close()
def start_data_processing(thread_number):
# open conn
conn = pymssql.connect(**agent_config_vars['mssql_kwargs'])
cursor = conn.cursor()
logger.info('Started connection number ' + str(thread_number))
sql_str = None
# get instance and metric mapping info
if agent_config_vars['instance_map_conn']:
try:
logger.info('Starting execute SQL to getting instance mapping info.')
sql_str = "select * from {}".format(agent_config_vars['instance_map_conn']['instance_map_table'])
logger.debug(sql_str)
# execute sql string
cursor.execute(sql_str)
instance_map = {}
password=self.cfg['password'], more=templates.connect(db=dbName) )
self.logSentEvent( shortEvt = "connect", tplEvt = self.encapsule(db_event=tpl) )
# connect
if self.cfg['agent-support']:
remote_cfg = { 'cmd': 'Connect', 'db-name': dbName , 'dbtype': 'mysql', 'user': self.cfg['user'],
'password':self.cfg['password'], 'host': self.cfg['host'], 'port': self.cfg['port'],
'timeout': int(timeout) }
self.sendNotifyToAgent(data=remote_cfg)
else:
try:
if pymssql.__version__ >= "2.0.0":
self.connPtr = pymssql.connect( host = self.cfg['host'], user = self.cfg['user'], password = self.cfg['password'],
port=self.cfg['port'], database = dbName, user_timeout=timeout )
else:
self.connPtr = pymssql.connect( host = self.cfg['host'], user = self.cfg['user'], password = self.cfg['password'],
database = dbName )
except Exception as e:
self.error( str(e) )
else:
# log connected event
tpl = templates.db( host=self.cfg['host'], port=self.cfg['port'], user=self.cfg['user'],
password=self.cfg['password'], more=templates.connected() )
self.logRecvEvent( shortEvt = "connected", tplEvt = self.encapsule(db_event=tpl) )
def connect(self):
self.initConnection()
try:
self.connector = pymssql.connect(host="%s:%d" % (self.hostname, self.port), user=self.user, password=self.password, database=self.db, login_timeout=conf.timeout, timeout=conf.timeout)
except (pymssql.InterfaceError, pymssql.OperationalError), msg:
raise SqlmapConnectionException(msg)
self.initCursor()
self.printConnected()
secret_dict (dict): The Secret Dictionary
Returns:
Connection: The pymssql.Connection object if successful. None otherwise
Raises:
KeyError: If the secret json does not contain the expected keys
"""
# Parse and validate the secret JSON string
port = str(secret_dict['port']) if 'port' in secret_dict else '1433'
dbname = secret_dict['dbname'] if 'dbname' in secret_dict else 'master'
# Try to obtain a connection to the db
try:
conn = pymssql.connect(server=secret_dict['host'],
user=secret_dict['username'],
password=secret_dict['password'],
database=dbname,
port=port,
login_timeout=5,
as_dict=True)
return conn
except pymssql.OperationalError:
return None
def Open(self, p_autocommit=True):
try:
self.v_con = pymssql.connect(
host=self.v_host,
port=int(self.v_port),
database=self.v_service,
user=self.v_user,
password=self.v_password
)
self.v_cur = self.v_con.cursor()
self.v_start = True
except pymssql.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
def Query(self, p_sql, p_alltypesstr=False, p_simple=False):