def _setup_environment(environ):
# Cygwin requires some special voodoo to set the environment variables
# properly so that Oracle will see them.
if platform.system().upper().startswith('CYGWIN'):
try:
import ctypes
except ImportError as e:
from django.core.exceptions import ImproperlyConfigured
raise ImproperlyConfigured("Error loading ctypes: %s; "
"the Oracle backend requires ctypes to "
"operate correctly under Cygwin." % e)
kernel32 = ctypes.CDLL('kernel32')
for name, value in environ:
kernel32.SetEnvironmentVariableA(name, value)
else:
os.environ.update(environ)
python类Error()的实例源码
def Execute(self, p_sql):
try:
v_keep = None
if self.v_con is None:
self.Open()
v_keep = False
else:
v_keep = True
self.v_cur.execute(p_sql)
except Spartacus.Database.Exception as exc:
raise exc
except sqlite3.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
finally:
if not v_keep:
self.Close()
def ExecuteScalar(self, p_sql):
try:
v_keep = None
if self.v_con is None:
self.Open()
v_keep = False
else:
v_keep = True
self.v_cur.execute(p_sql)
r = self.v_cur.fetchone()
if r != None:
s = r[0]
else:
s = None
return s
except Spartacus.Database.Exception as exc:
raise exc
except sqlite3.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
finally:
if not v_keep:
self.Close()
def InsertBlock(self, p_block, p_tablename, p_fields=None):
try:
v_columnames = []
if p_fields is None:
v_fields = []
for c in p_block.Columns:
v_columnames.append(c)
v_fields.append(DataField(c))
else:
v_fields = p_fields
for p in v_fields:
v_columnames.append(p.v_name)
v_insert = 'begin; '
for r in p_block.Rows:
v_insert = v_insert + 'insert into ' + p_tablename + '(' + ','.join(v_columnames) + ') values ' + self.Mogrify(r, v_fields) + '; '
v_insert = v_insert + 'commit;'
self.Execute(v_insert)
except Spartacus.Database.Exception as exc:
raise exc
except sqlite3.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
def ExecuteScalar(self, p_sql):
try:
if self.v_con is None:
raise Spartacus.Database.Exception('This method should be called in the middle of Open() and Close() calls.')
else:
self.v_cur.execute(p_sql)
r = self.v_cur.fetchone()
if r != None:
s = r[0]
else:
s = None
return s
except Spartacus.Database.Exception as exc:
raise exc
except sqlite3.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
def GetFields(self, p_sql):
try:
if self.v_con is None:
raise Spartacus.Database.Exception('This method should be called in the middle of Open() and Close() calls.')
else:
v_fields = []
self.v_cur.execute('select * from ( ' + p_sql + ' ) t limit 1')
r = self.v_cur.fetchone()
if r != None:
k = 0
for c in self.v_cur.description:
v_fields.append(DataField(c[0], p_type=type(r[k]), p_dbtype=type(r[k])))
k = k + 1
else:
k = 0
for c in self.v_cur.description:
v_fields.append(DataField(c[0], p_type=type(None), p_dbtype=type(None)))
k = k + 1
return v_fields
except Spartacus.Database.Exception as exc:
raise exc
except sqlite3.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
def Open(self, p_autocommit=True):
try:
self.v_con = psycopg2.connect(
self.GetConnectionString(),
cursor_factory=psycopg2.extras.DictCursor
)
self.v_con.autocommit = p_autocommit
self.v_cur = self.v_con.cursor()
self.v_start = True
# PostgreSQL types
self.v_cur.execute('select oid, typname from pg_type')
self.v_types = dict([(r['oid'], r['typname']) for r in self.v_cur.fetchall()])
if not p_autocommit:
self.v_con.commit()
self.v_con.notices = DataList()
except Spartacus.Database.Exception as exc:
raise exc
except psycopg2.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
def Execute(self, p_sql):
try:
v_keep = None
if self.v_con is None:
self.Open()
v_keep = False
else:
v_keep = True
self.v_cur.execute(p_sql)
except Spartacus.Database.Exception as exc:
raise exc
except psycopg2.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
finally:
if not v_keep:
self.Close()
def ExecuteScalar(self, p_sql):
try:
v_keep = None
if self.v_con is None:
self.Open()
v_keep = False
else:
v_keep = True
self.v_cur.execute(p_sql)
r = self.v_cur.fetchone()
if r != None:
s = r[0]
else:
s = None
return s
except Spartacus.Database.Exception as exc:
raise exc
except psycopg2.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
finally:
if not v_keep:
self.Close()
def InsertBlock(self, p_block, p_tablename, p_fields=None):
try:
v_columnames = []
if p_fields is None:
v_fields = []
for c in p_block.Columns:
v_columnames.append(c)
v_fields.append(DataField(c))
else:
v_fields = p_fields
for p in v_fields:
v_columnames.append(p.v_name)
v_values = []
for r in p_block.Rows:
v_values.append(self.Mogrify(r, v_fields))
self.Execute('insert into ' + p_tablename + '(' + ','.join(v_columnames) + ') values ' + ','.join(v_values) + '')
except Spartacus.Database.Exception as exc:
raise exc
except psycopg2.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
def Execute(self, p_sql):
try:
v_keep = None
if self.v_con is None:
self.Open()
v_keep = False
else:
v_keep = True
self.v_cur.execute(p_sql)
except Spartacus.Database.Exception as exc:
raise exc
except pymysql.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
finally:
if not v_keep:
self.Close()
def InsertBlock(self, p_block, p_tablename, p_fields=None):
try:
v_columnames = []
if p_fields is None:
v_fields = []
for c in p_block.Columns:
v_columnames.append(c)
v_fields.append(DataField(c))
else:
v_fields = p_fields
for p in v_fields:
v_columnames.append(p.v_name)
v_values = []
for r in p_block.Rows:
v_values.append(self.Mogrify(r, v_fields))
self.Execute('insert into ' + p_tablename + '(' + ','.join(v_columnames) + ') values ' + ','.join(v_values) + '')
except Spartacus.Database.Exception as exc:
raise exc
except pymysql.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
def Execute(self, p_sql):
try:
v_keep = None
if self.v_con is None:
self.Open()
v_keep = False
else:
v_keep = True
self.v_cur.execute(p_sql)
except Spartacus.Database.Exception as exc:
raise exc
except pymysql.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
finally:
if not v_keep:
self.Close()
def ExecuteScalar(self, p_sql):
try:
v_keep = None
if self.v_con is None:
self.Open()
v_keep = False
else:
v_keep = True
self.v_cur.execute(p_sql)
r = self.v_cur.fetchone()
if r != None:
s = r[0]
else:
s = None
return s
except Spartacus.Database.Exception as exc:
raise exc
except pymysql.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
finally:
if not v_keep:
self.Close()
def InsertBlock(self, p_block, p_tablename, p_fields=None):
try:
v_columnames = []
if p_fields is None:
v_fields = []
for c in p_block.Columns:
v_columnames.append(c)
v_fields.append(DataField(c))
else:
v_fields = p_fields
for p in v_fields:
v_columnames.append(p.v_name)
v_values = []
for r in p_block.Rows:
v_values.append(self.Mogrify(r, v_fields))
self.Execute('insert into ' + p_tablename + '(' + ','.join(v_columnames) + ') values ' + ','.join(v_values) + '')
except Spartacus.Database.Exception as exc:
raise exc
except pymysql.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
def Execute(self, p_sql):
try:
v_keep = None
if self.v_con is None:
self.Open()
v_keep = False
else:
v_keep = True
self.v_cur.execute(p_sql)
except Spartacus.Database.Exception as exc:
raise exc
except fdb.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
finally:
if not v_keep:
self.Close()
def InsertBlock(self, p_block, p_tablename, p_fields=None):
try:
v_columnames = []
if p_fields is None:
v_fields = []
for c in p_block.Columns:
v_columnames.append(c)
v_fields.append(DataField(c))
else:
v_fields = p_fields
for p in v_fields:
v_columnames.append(p.v_name)
v_values = []
for r in p_block.Rows:
v_values.append(self.Mogrify(r, v_fields))
self.Execute('insert into ' + p_tablename + '(' + ','.join(v_columnames) + ') values ' + ','.join(v_values) + '')
except Spartacus.Database.Exception as exc:
raise exc
except fdb.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
def Execute(self, p_sql):
try:
v_keep = None
if self.v_con is None:
self.Open()
v_keep = False
else:
v_keep = True
self.v_cur.execute(p_sql)
except Spartacus.Database.Exception as exc:
raise exc
except cx_Oracle.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
finally:
if not v_keep:
self.Close()
def ExecuteScalar(self, p_sql):
try:
v_keep = None
if self.v_con is None:
self.Open()
v_keep = False
else:
v_keep = True
self.v_cur.execute(p_sql)
r = self.v_cur.fetchone()
if r != None:
s = r[0]
else:
s = None
return s
except Spartacus.Database.Exception as exc:
raise exc
except cx_Oracle.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
finally:
if not v_keep:
self.Close()
def InsertBlock(self, p_block, p_tablename, p_fields=None):
try:
v_columnames = []
if p_fields is None:
v_fields = []
for c in p_block.Columns:
v_columnames.append(c)
v_fields.append(DataField(c))
else:
v_fields = p_fields
for p in v_fields:
v_columnames.append(p.v_name)
v_values = []
for r in p_block.Rows:
v_values.append(self.Mogrify(r, v_fields))
self.Execute('insert into ' + p_tablename + '(' + ','.join(v_columnames) + ') values ' + ','.join(v_values) + '')
except Spartacus.Database.Exception as exc:
raise exc
except cx_Oracle.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
def Execute(self, p_sql):
try:
v_keep = None
if self.v_con is None:
self.Open()
v_keep = False
else:
v_keep = True
self.v_cur.execute(p_sql)
except Spartacus.Database.Exception as exc:
raise exc
except pymssql.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
finally:
if not v_keep:
self.Close()
def InsertBlock(self, p_block, p_tablename, p_fields=None):
try:
v_columnames = []
if p_fields is None:
v_fields = []
for c in p_block.Columns:
v_columnames.append(c)
v_fields.append(DataField(c))
else:
v_fields = p_fields
for p in v_fields:
v_columnames.append(p.v_name)
v_values = []
for r in p_block.Rows:
v_values.append(self.Mogrify(r, v_fields))
self.Execute('insert into ' + p_tablename + '(' + ','.join(v_columnames) + ') values ' + ','.join(v_values) + '')
except Spartacus.Database.Exception as exc:
raise exc
except pymssql.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
def Execute(self, p_sql):
try:
v_keep = None
if self.v_con is None:
self.Open()
v_keep = False
else:
v_keep = True
self.v_cur.execute(p_sql)
except Spartacus.Database.Exception as exc:
raise exc
except ibm_db_dbi.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
finally:
if not v_keep:
self.Close()
def ExecuteScalar(self, p_sql):
try:
v_keep = None
if self.v_con is None:
self.Open()
v_keep = False
else:
v_keep = True
self.v_cur.execute(p_sql)
r = self.v_cur.fetchone()
if r != None:
s = r[0]
else:
s = None
return s
except Spartacus.Database.Exception as exc:
raise exc
except ibm_db_dbi.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
finally:
if not v_keep:
self.Close()
def InsertBlock(self, p_block, p_tablename, p_fields=None):
try:
v_columnames = []
if p_fields is None:
v_fields = []
for c in p_block.Columns:
v_columnames.append(c)
v_fields.append(DataField(c))
else:
v_fields = p_fields
for p in v_fields:
v_columnames.append(p.v_name)
v_values = []
for r in p_block.Rows:
v_values.append(self.Mogrify(r, v_fields))
self.Execute('insert into ' + p_tablename + '(' + ','.join(v_columnames) + ') values ' + ','.join(v_values) + '')
except Spartacus.Database.Exception as exc:
raise exc
except ibm_db_dbi.Error as exc:
raise Spartacus.Database.Exception(str(exc))
except Exception as exc:
raise Spartacus.Database.Exception(str(exc))
def _setup_environment(environ):
# Cygwin requires some special voodoo to set the environment variables
# properly so that Oracle will see them.
if platform.system().upper().startswith('CYGWIN'):
try:
import ctypes
except ImportError as e:
from django.core.exceptions import ImproperlyConfigured
raise ImproperlyConfigured("Error loading ctypes: %s; "
"the Oracle backend requires ctypes to "
"operate correctly under Cygwin." % e)
kernel32 = ctypes.CDLL('kernel32')
for name, value in environ:
kernel32.SetEnvironmentVariableA(name, value)
else:
os.environ.update(environ)
def _setup_environment(environ):
# Cygwin requires some special voodoo to set the environment variables
# properly so that Oracle will see them.
if platform.system().upper().startswith('CYGWIN'):
try:
import ctypes
except ImportError as e:
from django.core.exceptions import ImproperlyConfigured
raise ImproperlyConfigured("Error loading ctypes: %s; "
"the Oracle backend requires ctypes to "
"operate correctly under Cygwin." % e)
kernel32 = ctypes.CDLL('kernel32')
for name, value in environ:
kernel32.SetEnvironmentVariableA(name, value)
else:
os.environ.update(environ)
def _setup_environment(environ):
# Cygwin requires some special voodoo to set the environment variables
# properly so that Oracle will see them.
if platform.system().upper().startswith('CYGWIN'):
try:
import ctypes
except ImportError as e:
from django.core.exceptions import ImproperlyConfigured
raise ImproperlyConfigured("Error loading ctypes: %s; "
"the Oracle backend requires ctypes to "
"operate correctly under Cygwin." % e)
kernel32 = ctypes.CDLL('kernel32')
for name, value in environ:
kernel32.SetEnvironmentVariableA(name, value)
else:
os.environ.update(environ)
def _setup_environment(environ):
# Cygwin requires some special voodoo to set the environment variables
# properly so that Oracle will see them.
if platform.system().upper().startswith('CYGWIN'):
try:
import ctypes
except ImportError as e:
from django.core.exceptions import ImproperlyConfigured
raise ImproperlyConfigured("Error loading ctypes: %s; "
"the Oracle backend requires ctypes to "
"operate correctly under Cygwin." % e)
kernel32 = ctypes.CDLL('kernel32')
for name, value in environ:
kernel32.SetEnvironmentVariableA(name, value)
else:
os.environ.update(environ)
def _setup_environment(environ):
# Cygwin requires some special voodoo to set the environment variables
# properly so that Oracle will see them.
if platform.system().upper().startswith('CYGWIN'):
try:
import ctypes
except ImportError as e:
from django.core.exceptions import ImproperlyConfigured
raise ImproperlyConfigured("Error loading ctypes: %s; "
"the Oracle backend requires ctypes to "
"operate correctly under Cygwin." % e)
kernel32 = ctypes.CDLL('kernel32')
for name, value in environ:
kernel32.SetEnvironmentVariableA(name, value)
else:
os.environ.update(environ)
