Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Simple MSSQL python interface - you'll need to build all of your actual insert/query functions (don't forget to commit on insert! ;D)
# Universally usable MSSQL interface Just modify the connection info and add specific module if necessary
import datetime, pyodbc
class MSSQLinterface():
# Init module
def __init__(self):
# Pulls global connection info into properties
self.dbUser = 'sa'
self.dbPass = 'password'
self.dbHost = '(local)'
self.dbPort = '1433'
self.driver = '{SQL Server}'
self.database = 'dbName'
self.connectionString = 'UID={0};PWD={1};DATABASE={2};PORT={3};SERVER={4};DRIVER={5}'.format(self.dbUser, self.dbPass, self.database, self.dbPort, self.dbHost, self.driver)
# Connection status and object properties
self.dbCursor = None
self.dbConnection = None
self.isConnected = False
self.connectedServer = None
self.queryText = None
# Logging directory and other variables for output control
self.logDirectory = 'C:\SQLinterface\Logs'
self.verboseOutput = True
self.outputEnabled = True
# Output control module
def printMsg(self, messageText, messageType, isLogged):
if isLogged is True:
f = open(self.logDirectory + '\\' + "{:%m-%d-%Y}".format(datetime.date.today()) + '.log', 'a')
f.write('{:%m-%d-%Y %H:%M:%S}'.format(datetime.datetime.now()) + ': ' + messageText + '\n')
f.close()
if self.outputEnabled is True:
if messageType == 0:
print('[+] ' + messageText)
if self.verboseOutput is True and messageType == 1:
print('[!] ' + messageText + '\n')
def sqlDataToDictionary(self, sqlData, cursor):
sqlDictList = []
columns = [column[0] for column in cursor.description]
for row in sqlData:
sqlDictList.append(dict(zip(columns, row)))
return sqlDictList
def connectSQL(self):
self.dbConnection = pyodbc.connect(self.connectionString)
connectedMessage = 'Connecting to database %s on server %s:%s (MSSSQL) as user %s' % (self.database, self.dbHost, self.dbPort, self.dbUser)
self.printMsg(connectedMessage, 0, True)
self.dbCursor = self.dbConnection.cursor()
self.isConnected = True
def execQuery(self, queryString):
self.dbCursor.execute(queryString)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.