Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def pyodbc_connect_and_query(odbc_connect_string: str, query: str) -> pyodbc.Cursor:
try:
conn = pyodbc.connect(odbc_connect_string)
conn.setdecoding(pyodbc.SQL_WCHAR, encoding='utf-8')
conn.setencoding(str, encoding='utf-8')
conn.setencoding(str, encoding='utf-8', ctype=pyodbc.SQL_CHAR)
# https://github.com/mkleehammer/pyodbc/issues/194 for this encoding fix
conn.setdecoding(pyodbc.SQL_WMETADATA, encoding='utf-32le')
except pyodbc.Error as e:
print("ODBC Connection Error Message:\n")
print(e)
print("ODBC error, exiting...\n")
sys.exit()
cursor = conn.cursor()
try:
cursor.execute(query)
except pyodbc.ProgrammingError as e:
print("\nODBC Query Error Message:\n")
print(e)
print("\nODBC error, exiting...\n")
sys.exit()
return cursor
if 'outputFile' in kw:
self.outputFileObject = kw['outputFile']
del kw['outputFile']
else:
self.outputFileObject = sys.stdout
if 'scriptSchema' in kw:
self.scriptSchema = kw['scriptSchema']
del kw['scriptSchema']
else:
self.scriptSchema = None
self.odbc = pyodbc.connect(*args, **kw)
if PY3:
self.odbc.setdecoding(pyodbc.SQL_CHAR, encoding='utf-8')
self.odbc.setdecoding(pyodbc.SQL_WCHAR, encoding='utf-8')
self.odbc.setdecoding(pyodbc.SQL_WMETADATA, encoding='utf-8')
self.odbc.setencoding(encoding='utf-8')
if self.serverAddress is None:
host, port = tuple(self.odbc.getinfo(pyodbc.SQL_SERVER_NAME).split(':'))
self.serverAddress = (str(host), int(port))
self.error = None
self._outputService = None
self._connected = True
if self.clientAddress is not None and not self.externalClient:
self._startOutputService()
self._outputLock = threading.Lock()
self._q = lambda x, q: q and '"%s"' % str(x).replace('"', '""') or str(x)
def initialize_connection(self):
connection = self._connection
if sys.version_info[0] < 3:
connection.setdecoding(pyodbc.SQL_CHAR, 'utf-8', pyodbc.SQL_CHAR)
connection.setdecoding(pyodbc.SQL_WCHAR, 'utf-32LE', pyodbc.SQL_WCHAR, unicode)
connection.setdecoding(pyodbc.SQL_WMETADATA, 'utf-32LE', pyodbc.SQL_WCHAR, unicode)
connection.setencoding(unicode, 'utf-32LE', pyodbc.SQL_WCHAR)
connection.setencoding(str, 'utf-8', pyodbc.SQL_CHAR)
else:
connection.setdecoding(pyodbc.SQL_CHAR, 'utf-8', pyodbc.SQL_CHAR)
connection.setdecoding(pyodbc.SQL_WCHAR, 'utf-32LE', pyodbc.SQL_WCHAR)
connection.setdecoding(pyodbc.SQL_WMETADATA, 'utf-32LE', pyodbc.SQL_WCHAR)
connection.setencoding('utf-32LE', pyodbc.SQL_WCHAR)
connection.setencoding('utf-8', pyodbc.SQL_CHAR)
self.__init_ns_decls__()
def connect(self, *args, **kwargs):
connection = super(VirtuosoDialect, self).connect(*args, **kwargs)
if util.py2k:
connection.setdecoding(pyodbc.SQL_CHAR, 'utf-8', pyodbc.SQL_CHAR)
connection.setdecoding(pyodbc.SQL_WCHAR, 'utf-32LE', pyodbc.SQL_WCHAR, unicode)
connection.setdecoding(pyodbc.SQL_WMETADATA, 'utf-32LE', pyodbc.SQL_WCHAR, unicode)
connection.setencoding(unicode, 'utf-32LE', pyodbc.SQL_WCHAR)
connection.setencoding(str, 'utf-8', pyodbc.SQL_CHAR)
else:
connection.setdecoding(pyodbc.SQL_CHAR, 'utf-8', pyodbc.SQL_CHAR)
connection.setdecoding(pyodbc.SQL_WCHAR, 'utf-32LE', pyodbc.SQL_WCHAR)
connection.setdecoding(pyodbc.SQL_WMETADATA, 'utf-32LE', pyodbc.SQL_WCHAR)
connection.setencoding('utf-32LE', pyodbc.SQL_WCHAR)
connection.setencoding('utf-8', pyodbc.SQL_CHAR)
return connection
def basicQuery(idNumber):
destination = "filemaker"
user = login(destination)[0]
cred = login(destination)[1]
# OPEN CONNECTION TO FILEMAKER DATABASE WITH DESCRIPTIVE METADATA
c = pyodbc.connect("DRIVER={FileMaker ODBC};DSN=pfacollection;SERVER=bampfa-pfm13.ist.1918.berkeley.edu;UID="+user+";PWD="+cred)
c.setdecoding(pyodbc.SQL_CHAR, encoding='utf-8')
c.setdecoding(pyodbc.SQL_WCHAR, encoding='utf-8')
c.setdecoding(pyodbc.SQL_WMETADATA, encoding='utf-8')
c.setencoding(encoding='utf-8')
cursor= c.cursor()
# SQL TO GET REQUIRED METADATA VALUES FROM FM
cursor.execute("""SELECT m_245a_CompleteTitle, AlternativeTitle,
AccessionNumberPrefix, AccessionNumberDepositorNumber,
AccessionNumberItemNumber, ProjectGroupTitle,
m_257a_Country, m_260c_ReleaseYear,
ct_DirectorsNames, Credits,
GeneralNotes, m_945z_GeneralConditionNotes
FROM CollectionItem WHERE AccessionNumberItemNumber = ?""",idNumber)
rows = cursor.fetchall()
resultData = {}
resultList = [x for y in rows for x in y]
# A NULL RESULT WILL YIELD AN EMPTY LIST, SO CHECK THAT SOMETHING WAS FOUND, OR SKIP THE FILE.
def connect(self, *args, **kwargs):
connection = super(VirtuosoDialect, self).connect(*args, **kwargs)
if util.py2k:
connection.setdecoding(pyodbc.SQL_CHAR, 'utf-8', pyodbc.SQL_CHAR)
connection.setdecoding(pyodbc.SQL_WCHAR, 'utf-32LE', pyodbc.SQL_WCHAR, unicode)
connection.setdecoding(pyodbc.SQL_WMETADATA, 'utf-32LE', pyodbc.SQL_WCHAR, unicode)
connection.setencoding(unicode, 'utf-32LE', pyodbc.SQL_WCHAR)
connection.setencoding(str, 'utf-8', pyodbc.SQL_CHAR)
else:
connection.setdecoding(pyodbc.SQL_CHAR, 'utf-8', pyodbc.SQL_CHAR)
connection.setdecoding(pyodbc.SQL_WCHAR, 'utf-32LE', pyodbc.SQL_WCHAR)
connection.setdecoding(pyodbc.SQL_WMETADATA, 'utf-32LE', pyodbc.SQL_WCHAR)
connection.setencoding('utf-32LE', pyodbc.SQL_WCHAR)
connection.setencoding('utf-8', pyodbc.SQL_CHAR)
return connection
def query(idNumber,basename):
# OPEN CONNECTION TO FILEMAKER DATABASE WITH DESCRIPTIVE METADATA
c = pyodbc.connect("DRIVER={FileMaker ODBC};DATABASE=PFA_Collection;SERVER=bampfa-pfm13.ist.1918.berkeley.edu;UID=resourcespace;PWD=mediaarchive2017")
c.setdecoding(pyodbc.SQL_CHAR, encoding='utf-8')
c.setdecoding(pyodbc.SQL_WCHAR, encoding='utf-8')
c.setdecoding(pyodbc.SQL_WMETADATA, encoding='utf-8')
c.setencoding(encoding='utf-8')
cursor= c.cursor()
# SQL TO GET REQUIRED METADATA VALUES FROM FM
# SQL TO GET REQUIRED METADATA VALUES FROM FM
cursor.execute("SELECT m_245a_CompleteTitle FROM CollectionItem WHERE AccessionNumberItemNumber = ?",idNumber)
# for row in cursor.tables():
# print(row.table_name)
rows = cursor.fetchall()
resultData = {}
resultList = [x for y in rows for x in y]
# print(resultList)
# A NULL RESULT WILL YIELD AN EMPTY LIST, SO CHECK THAT SOMETHING WAS FOUND, OR SKIP THE FILE.
# IF I SET UP LOGGING THIS SHOULD BE LOGGED FOR SURE.
def query(idNumber,basename):
# OPEN CONNECTION TO FILEMAKER DATABASE WITH DESCRIPTIVE METADATA
c = pyodbc.connect("DRIVER={FileMaker ODBC};DSN=mmTest;SERVER=localhost;UID=michael;PWD=michael")
c.setdecoding(pyodbc.SQL_CHAR, encoding='utf-8')
c.setdecoding(pyodbc.SQL_WCHAR, encoding='utf-8')
c.setdecoding(pyodbc.SQL_WMETADATA, encoding='utf-8')
c.setencoding(encoding='utf-8')
cursor= c.cursor()
# SQL TO GET REQUIRED METADATA VALUES FROM FM
cursor.execute("""
SELECT summary, tilte FROM microservicesTrial WHERE original_accession_number = ?
""",idNumber)
rows = cursor.fetchall()
resultData = {}
resultList = [x for y in rows for x in y]
# print(resultList)
# A NULL RESULT WILL YIELD AN EMPTY LIST, SO CHECK THAT SOMETHING WAS FOUND, OR SKIP THE FILE.