0001from sqlobject.dbconnection import DBAPI
0002from sqlobject import col
0003Sybase = None
0004
0005class SybaseConnection(DBAPI):
0006
0007 supportTransactions = False
0008 dbName = 'sybase'
0009 schemes = [dbName]
0010
0011 def __init__(self, db, user, passwd='', host='localhost',
0012 locking=1, **kw):
0013 db = db.strip('/')
0014 global Sybase
0015 if Sybase is None:
0016 import Sybase
0017 Sybase._ctx.debug = 0
0018 from Sybase import NumericType
0019 from sqlobject.converters import registerConverter, IntConverter
0020 registerConverter(NumericType, IntConverter)
0021 self.module = Sybase
0022 self.locking = int(locking)
0023 self.host = host
0024 self.db = db
0025 self.user = user
0026 self.passwd = passwd
0027 autoCommit = kw.get('autoCommit')
0028 if autoCommit:
0029 autoCommmit = int(autoCommit)
0030 else:
0031 autoCommit = None
0032 kw['autoCommit'] = autoCommit
0033 DBAPI.__init__(self, **kw)
0034
0035 def connectionFromURI(cls, uri):
0036 user, password, host, port, path, args = cls._parseURI(uri)
0037 return cls(user=user, passwd=password, host=host or 'localhost',
0038 db=path, **args)
0039 connectionFromURI = classmethod(connectionFromURI)
0040
0041 def insert_id(self, conn):
0042 """
0043 Sybase adapter/cursor does not support the
0044 insert_id method.
0045 """
0046 c = conn.cursor()
0047 c.execute('SELECT @@IDENTITY')
0048 return c.fetchone()[0]
0049
0050 def makeConnection(self):
0051 return Sybase.connect(self.host, self.user, self.passwd,
0052 database=self.db, auto_commit=self.autoCommit,
0053 locking=self.locking)
0054
0055 HAS_IDENTITY = """
0056 SELECT col.name, col.status, obj.name
0057 FROM syscolumns col
0058 JOIN sysobjects obj
0059 ON obj.id = col.id
0060 WHERE obj.name = '%s'
0061 AND (col.status & 0x80) = 0x80
0062 """
0063 def _hasIdentity(self, conn, table):
0064 query = self.HAS_IDENTITY % table
0065 c = conn.cursor()
0066 c.execute(query)
0067 r = c.fetchone()
0068 return r is not None
0069
0070 def _queryInsertID(self, conn, soInstance, id, names, values):
0071 table = soInstance.sqlmeta.table
0072 idName = soInstance.sqlmeta.idName
0073 c = conn.cursor()
0074 if id is not None:
0075 names = [idName] + names
0076 values = [id] + values
0077
0078 has_identity = self._hasIdentity(conn, table)
0079 if has_identity:
0080 if id is not None:
0081 c.execute('SET IDENTITY_INSERT %s ON' % table)
0082 else:
0083 c.execute('SET IDENTITY_INSERT %s OFF' % table)
0084
0085 q = self._insertSQL(table, names, values)
0086 if self.debug:
0087 print 'QueryIns: %s' % q
0088 c.execute(q)
0089 if has_identity:
0090 c.execute('SET IDENTITY_INSERT %s OFF' % table)
0091 if id is None:
0092 id = self.insert_id(conn)
0093 if self.debugOutput:
0094 self.printDebug(conn, id, 'QueryIns', 'result')
0095 return id
0096
0097 def _queryAddLimitOffset(self, query, start, end):
0098
0099 return query
0100
0101 def createColumn(self, soClass, col):
0102 return col.sybaseCreateSQL()
0103
0104 def createIDColumn(self, soClass):
0105 return '%s NUMERIC(18,0) IDENTITY UNIQUE' % soClass.sqlmeta.idName
0106
0107 def createIndexSQL(self, soClass, index):
0108 return index.sybaseCreateIndexSQL(soClass)
0109
0110 def joinSQLType(self, join):
0111 return 'NUMERIC(18,0) NOT NULL'
0112
0113 SHOW_TABLES="SELECT name FROM sysobjects WHERE type='U'"
0114 def tableExists(self, tableName):
0115 for (table,) in self.queryAll(self.SHOW_TABLES):
0116 if table.lower() == tableName.lower():
0117 return True
0118 return False
0119
0120 def addColumn(self, tableName, column):
0121 self.query('ALTER TABLE %s ADD COLUMN %s' %
0122 (tableName,
0123 column.sybaseCreateSQL()))
0124
0125 def delColumn(self, tableName, column):
0126 self.query('ALTER TABLE %s DROP COLUMN %s' %
0127 (tableName,
0128 column.dbName))
0129
0130 SHOW_COLUMNS=('SELECT COLUMN_NAME, DATA_TYPE, IS_NULLABLE, COLUMN_DEFAULT FROM INFORMATION_SCHEMA.COLUMNS '
0131 'WHERE TABLE_NAME = \'%s\'')
0132 def columnsFromSchema(self, tableName, soClass):
0133 colData = self.queryAll(self.SHOW_COLUMNS
0134 % tableName)
0135 results = []
0136 for field, t, nullAllowed, default in colData:
0137 if field == 'id':
0138 continue
0139 colClass, kw = self.guessClass(t)
0140 kw['name'] = soClass.sqlmeta.style.dbColumnToPythonAttr(field)
0141 kw['notNone'] = not nullAllowed
0142 kw['default'] = default
0143
0144
0145 kw['forceDBName'] = True
0146 results.append(colClass(**kw))
0147 return results
0148
0149 def _setAutoCommit(self, conn, auto):
0150 conn.auto_commit = auto
0151
0152 def guessClass(self, t):
0153 if t.startswith('int'):
0154 return col.IntCol, {}
0155 elif t.startswith('varchar'):
0156 return col.StringCol, {'length': int(t[8:-1])}
0157 elif t.startswith('char'):
0158 return col.StringCol, {'length': int(t[5:-1]),
0159 'varchar': False}
0160 elif t.startswith('datetime'):
0161 return col.DateTimeCol, {}
0162 else:
0163 return col.Col, {}