/usr/share/pyshared/sqlobject/sybase/sybaseconnection.py is in python-sqlobject 0.12.4-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 | from sqlobject.dbconnection import DBAPI
from sqlobject import col
class SybaseConnection(DBAPI):
supportTransactions = False
dbName = 'sybase'
schemes = [dbName]
NumericType = None
def __init__(self, db, user, password='', host='localhost',
locking=1, **kw):
db = db.strip('/')
import Sybase
Sybase._ctx.debug = 0
if SybaseConnection.NumericType is None:
from Sybase import NumericType
SybaseConnection.NumericType = NumericType
from sqlobject.converters import registerConverter, IntConverter
registerConverter(NumericType, IntConverter)
self.module = Sybase
self.locking = int(locking)
self.host = host
self.db = db
self.user = user
self.password = password
autoCommit = kw.get('autoCommit')
if autoCommit:
autoCommmit = int(autoCommit)
else:
autoCommit = None
kw['autoCommit'] = autoCommit
DBAPI.__init__(self, **kw)
def connectionFromURI(cls, uri):
user, password, host, port, path, args = cls._parseURI(uri)
return cls(user=user, password=password, host=host or 'localhost',
db=path, **args)
connectionFromURI = classmethod(connectionFromURI)
def insert_id(self, conn):
"""
Sybase adapter/cursor does not support the
insert_id method.
"""
c = conn.cursor()
c.execute('SELECT @@IDENTITY')
return c.fetchone()[0]
def makeConnection(self):
return self.module.connect(self.host, self.user, self.password,
database=self.db, auto_commit=self.autoCommit,
locking=self.locking)
HAS_IDENTITY = """
SELECT col.name, col.status, obj.name
FROM syscolumns col
JOIN sysobjects obj
ON obj.id = col.id
WHERE obj.name = '%s'
AND (col.status & 0x80) = 0x80
"""
def _hasIdentity(self, conn, table):
query = self.HAS_IDENTITY % table
c = conn.cursor()
c.execute(query)
r = c.fetchone()
return r is not None
def _queryInsertID(self, conn, soInstance, id, names, values):
table = soInstance.sqlmeta.table
idName = soInstance.sqlmeta.idName
c = conn.cursor()
if id is not None:
names = [idName] + names
values = [id] + values
has_identity = self._hasIdentity(conn, table)
identity_insert_on = False
if has_identity and (id is not None):
identity_insert_on = True
c.execute('SET IDENTITY_INSERT %s ON' % table)
q = self._insertSQL(table, names, values)
if self.debug:
print 'QueryIns: %s' % q
c.execute(q)
if has_identity and identity_insert_on:
c.execute('SET IDENTITY_INSERT %s OFF' % table)
if id is None:
id = self.insert_id(conn)
if self.debugOutput:
self.printDebug(conn, id, 'QueryIns', 'result')
return id
def _queryAddLimitOffset(cls, query, start, end):
# XXX Sybase doesn't support OFFSET
if end:
return "SET ROWCOUNT %i %s SET ROWCOUNT 0" % (end, query)
return query
_queryAddLimitOffset = classmethod(_queryAddLimitOffset)
def createReferenceConstraint(self, soClass, col):
return None
def createColumn(self, soClass, col):
return col.sybaseCreateSQL()
def createIDColumn(self, soClass):
key_type = {int: "NUMERIC(18,0)", str: "TEXT"}[soClass.sqlmeta.idType]
return '%s %s IDENTITY UNIQUE' % (soClass.sqlmeta.idName, key_type)
def createIndexSQL(self, soClass, index):
return index.sybaseCreateIndexSQL(soClass)
def joinSQLType(self, join):
return 'NUMERIC(18,0) NOT NULL'
SHOW_TABLES="SELECT name FROM sysobjects WHERE type='U'"
def tableExists(self, tableName):
for (table,) in self.queryAll(self.SHOW_TABLES):
if table.lower() == tableName.lower():
return True
return False
def addColumn(self, tableName, column):
self.query('ALTER TABLE %s ADD COLUMN %s' %
(tableName,
column.sybaseCreateSQL()))
def delColumn(self, sqlmeta, column):
self.query('ALTER TABLE %s DROP COLUMN %s' % (sqlmeta.table, column.dbName))
SHOW_COLUMNS=('SELECT COLUMN_NAME, DATA_TYPE, IS_NULLABLE, COLUMN_DEFAULT FROM INFORMATION_SCHEMA.COLUMNS '
'WHERE TABLE_NAME = \'%s\'')
def columnsFromSchema(self, tableName, soClass):
colData = self.queryAll(self.SHOW_COLUMNS
% tableName)
results = []
for field, t, nullAllowed, default in colData:
if field == soClass.sqlmeta.idName:
continue
colClass, kw = self.guessClass(t)
kw['name'] = soClass.sqlmeta.style.dbColumnToPythonAttr(field)
kw['dbName'] = field
kw['notNone'] = not nullAllowed
kw['default'] = default
# @@ skip key...
# @@ skip extra...
kw['forceDBName'] = True
results.append(colClass(**kw))
return results
def _setAutoCommit(self, conn, auto):
conn.auto_commit = auto
def guessClass(self, t):
if t.startswith('int'):
return col.IntCol, {}
elif t.startswith('varchar'):
return col.StringCol, {'length': int(t[8:-1])}
elif t.startswith('char'):
return col.StringCol, {'length': int(t[5:-1]),
'varchar': False}
elif t.startswith('datetime'):
return col.DateTimeCol, {}
else:
return col.Col, {}
|