- many incantations to get the tests to run reasonably

- executemany() for some reason uses some tiny buffer, overriding it
- we need to use the IDENTITY_INSERT thing
This commit is contained in:
Mike Bayer 2010-03-14 23:03:24 +00:00
parent 39fd3442e3
commit 127c02747c
3 changed files with 70 additions and 17 deletions

View File

@ -124,16 +124,30 @@ Additional steps specific to individual databases are as follows:
grant dba to scott;
SYBASE: Similar to Oracle, "test_schema" is created as a user, and the
primary test user needs to have the "sa_role":
create database sqlalchemy
sp_addlogin scott, "tiger7"
sp_addlogin test_schema, "tiger7"
sp_adduser scott
sp_adduser test_schema
grant all to scott
grant sa_role to scott
primary test user needs to have the "sa_role".
It's also recommened to turn on "trunc log on chkpt" and to use a
separate transaction log device - Sybase basically seizes up when
the transaction log is full otherwise.
A full series of setup assuming sa/master:
disk init name="translog", physname="/opt/sybase/data/translog.dat", size="10M"
create database sqlalchemy on default log on translog="10M"
sp_dboption sqlalchemy, "trunc log on chkpt", true
sp_addlogin scott, "tiger7"
sp_addlogin test_schema, "tiger7"
use sqlalchemy
sp_adduser scott
sp_adduser test_schema
grant all to scott
sp_role "grant", sa_role, scott
Sybase will still freeze for up to a minute when the log becomes
full. To manually dump the log:
dump tran sqlalchemy with truncate_only
MSSQL: Tests that involve multiple connections require Snapshot Isolation
ability implented on the test database in order to prevent deadlocks that
will occur with record locking isolation. This feature is only available

View File

@ -13,7 +13,7 @@ ASE is the primary support platform.
"""
import operator
from sqlalchemy.sql import compiler, expression
from sqlalchemy.sql import compiler, expression, text, bindparam
from sqlalchemy.engine import default, base, reflection
from sqlalchemy import types as sqltypes
from sqlalchemy.sql import operators as sql_operators
@ -23,7 +23,7 @@ from sqlalchemy import util, sql, exc
from sqlalchemy.types import CHAR, VARCHAR, TIME, NCHAR, NVARCHAR,\
TEXT,DATE,DATETIME, FLOAT, NUMERIC,\
BIGINT,INT, INTEGER, SMALLINT, BINARY,\
VARBINARY
VARBINARY, DECIMAL, TIMESTAMP, Unicode
RESERVED_WORDS = set([
"add", "all", "alter", "and",
@ -175,12 +175,38 @@ ischema_names = {
class SybaseExecutionContext(default.DefaultExecutionContext):
def post_exec(self):
if self.isinsert and not self.executemany:
self.cursor.execute("SELECT @@identity AS lastrowid")
row = self.cursor.fetchall()[0]
self._lastrowid = int(row[0])
_enable_identity_insert = False
def pre_exec(self):
if self.isinsert:
tbl = self.compiled.statement.table
seq_column = tbl._autoincrement_column
insert_has_sequence = seq_column is not None
if insert_has_sequence:
self._enable_identity_insert = seq_column.key in self.compiled_parameters[0]
else:
self._enable_identity_insert = False
if self._enable_identity_insert:
self.cursor.execute("SET IDENTITY_INSERT %s ON" %
self.dialect.identifier_preparer.format_table(tbl))
def post_exec(self):
if self._enable_identity_insert:
self.cursor.execute(
"SET IDENTITY_INSERT %s OFF" %
self.dialect.identifier_preparer.
format_table(self.compiled.statement.table)
)
def get_lastrowid(self):
cursor = self.create_cursor()
cursor.execute("SELECT @@identity AS lastrowid")
lastrowid = cursor.fetchone()[0]
cursor.close()
return lastrowid
class SybaseSQLCompiler(compiler.SQLCompiler):
@ -301,6 +327,11 @@ class SybaseDialect(default.DefaultDialect):
supports_unicode_statements = False
supports_sane_rowcount = False
supports_sane_multi_rowcount = False
supports_native_boolean = False
supports_unicode_binds = False
postfetch_lastrowid = True
colspecs = colspecs
ischema_names = ischema_names

View File

@ -21,6 +21,8 @@ from sqlalchemy.dialects.sybase.base import SybaseDialect, \
class SybaseExecutionContext_pysybase(SybaseExecutionContext):
def pre_exec(self):
SybaseExecutionContext.pre_exec(self)
for param in self.parameters:
for key in list(param):
param["@" + key] = param[key]
@ -58,6 +60,12 @@ class SybaseDialect_pysybase(SybaseDialect):
return ([opts.pop('host')], opts)
def do_executemany(self, cursor, statement, parameters, context=None):
# calling python-sybase executemany yields:
# TypeError: string too long for buffer
for param in parameters:
cursor.execute(statement, param)
def _get_server_version_info(self, connection):
return connection.scalar("select @@version_number")