from sqlalchemy import types as sqltypes
from decimal import Decimal as _python_Decimal
-import information_schema as ischema
+from sqlalchemy.dialects.mssql import information_schema as ischema
MS_2008_VERSION = (10,)
MS_2005_VERSION = (9,)
self._autorollback()
if self.__close_with_result:
self.close()
+ # Py3K
+ #raise exc.DBAPIError.instance(statement, parameters, e, connection_invalidated=is_disconnect).with_traceback(sys.exc_info()[2])
+ # Py2K
raise exc.DBAPIError.instance(statement, parameters, e, connection_invalidated=is_disconnect), None, sys.exc_info()[2]
+ # end Py2K
+
finally:
del self._reentrant_error
dialect = self.conn.dialect
- # for some work arounds
- from sqlalchemy.dialects.mysql.mysqldb import MySQLDialect
-
# MySQL dialect does this. Applicable with other dialects?
if hasattr(dialect, '_connection_charset') \
and hasattr(dialect, '_adjust_casing'):
del tblkw[k]
tblkw[str(k)] = v
- ### Py2K
+ # Py2K
if isinstance(schema, str):
schema = schema.decode(dialect.encoding)
if isinstance(table_name, str):
return dialect.connect(*cargs, **cparams)
except Exception, e:
import sys
+
+ # Py3K
+ #raise exc.DBAPIError.instance(None, None, e).with_traceback(sys.exc_info()[2])
+ # Py2K
raise exc.DBAPIError.instance(None, None, e), None, sys.exc_info()[2]
+ # end Py2K
+
creator = kwargs.pop('creator', connect)
poolclass = (kwargs.pop('poolclass', None) or
tokens = components['database'].split('?', 2)
components['database'] = tokens[0]
query = (len(tokens) > 1 and dict(cgi.parse_qsl(tokens[1]))) or None
+ # Py2K
if query is not None:
query = dict((k.encode('ascii'), query[k]) for k in query)
+ # end Py2K
else:
query = None
components['query'] = query
return lambda ctx: fn()
positionals = len(argspec[0])
- if inspect.ismethod(inspectable):
+
+ # Py3K compat - no unbound methods
+ if inspect.ismethod(inspectable) or inspect.isclass(fn):
positionals -= 1
if positionals == 0:
from sqlalchemy.sql import table, column
from sqlalchemy.databases import mssql
from sqlalchemy.dialects.mssql import pyodbc
-import sqlalchemy.engine.url as url
+from sqlalchemy.engine import url
from testlib import *
from testlib.testing import eq_
import testenv; testenv.configure_for_tests()
-import ConfigParser, StringIO
+import ConfigParser
+import StringIO
import sqlalchemy.engine.url as url
from sqlalchemy import create_engine, engine_from_config
import testlib.sa as tsa