from .. import util
+
def _auto_fn(name):
"""default dialect importer.
self.charset = charset
super(_StringType, self).__init__(**kw)
+
class VARCHAR(_StringType, sqltypes.VARCHAR):
"""Firebird VARCHAR type"""
__visit_name__ = 'VARCHAR'
def __init__(self, length=None, **kwargs):
super(VARCHAR, self).__init__(length=length, **kwargs)
+
class CHAR(_StringType, sqltypes.CHAR):
"""Firebird CHAR type"""
__visit_name__ = 'CHAR'
from .kinterbasdb import FBDialect_kinterbasdb
from ... import util
+
class FBDialect_fdb(FBDialect_kinterbasdb):
@classmethod
return value
return process
+
class FBExecutionContext_kinterbasdb(FBExecutionContext):
@property
def rowcount(self):
else:
return -1
+
class FBDialect_kinterbasdb(FBDialect):
driver = 'kinterbasdb'
supports_sane_rowcount = False
"xadatasource", "xid", "xload", "xunload", "year"
])
+
class InfoDateTime(sqltypes.DateTime):
+
def bind_processor(self, dialect):
def process(value):
if value is not None:
return value
return process
+
class InfoTime(sqltypes.Time):
+
def bind_processor(self, dialect):
def process(value):
if value is not None:
def visit_boolean(self, type_):
return "SMALLINT"
+
class InfoSQLCompiler(compiler.SQLCompiler):
+
def default_from(self):
return " from systables where tabname = 'systables' "
text += "CONSTRAINT %s " % self.preparer.format_constraint(constraint)
return text
+
class InformixIdentifierPreparer(compiler.IdentifierPreparer):
reserved_words = RESERVED_WORDS
VERSION_RE = re.compile(r'(\d+)\.(\d+)(.+\d+)')
+
class InformixExecutionContext_informixdb(default.DefaultExecutionContext):
+
def post_exec(self):
if self.isinsert:
self._lastrowid = self.cursor.sqlerrd[1]
from sqlalchemy.dialects.mssql.base import MSDateTime, MSDialect
import sys
+
class MSDateTime_adodbapi(MSDateTime):
def result_processor(self, dialect, coltype):
def process(value):
'writetext',
])
+
class REAL(sqltypes.REAL):
__visit_name__ = 'REAL'
kw['precision'] = 24
super(REAL, self).__init__(**kw)
+
class TINYINT(sqltypes.Integer):
__visit_name__ = 'TINYINT'
return value
return process
+
class TIME(sqltypes.TIME):
def __init__(self, precision=None, **kwargs):
self.precision = precision
return process
_MSTime = TIME
+
class _DateTimeBase(object):
def bind_processor(self, dialect):
def process(value):
return value
return process
+
class _MSDateTime(_DateTimeBase, sqltypes.DateTime):
pass
+
class SMALLDATETIME(_DateTimeBase, sqltypes.DateTime):
__visit_name__ = 'SMALLDATETIME'
+
class DATETIME2(_DateTimeBase, sqltypes.DateTime):
__visit_name__ = 'DATETIME2'
def __init__(self, precision=None, **kwargs):
self.precision = precision
+
class _StringType(object):
"""Base for MSSQL string types."""
def __init__(self, collation=None):
self.collation = collation
+
class TEXT(_StringType, sqltypes.TEXT):
"""MSSQL TEXT type, for variable-length text up to 2^31 characters."""
_StringType.__init__(self, collation)
sqltypes.Text.__init__(self, length, **kw)
+
class NTEXT(_StringType, sqltypes.UnicodeText):
"""MSSQL NTEXT type, for variable-length unicode text up to 2^30
characters."""
_StringType.__init__(self, collation)
sqltypes.VARCHAR.__init__(self, length, **kw)
+
class NVARCHAR(_StringType, sqltypes.NVARCHAR):
"""MSSQL NVARCHAR type.
_StringType.__init__(self, collation)
sqltypes.NVARCHAR.__init__(self, length, **kw)
+
class CHAR(_StringType, sqltypes.CHAR):
"""MSSQL CHAR type, for fixed-length non-Unicode data with a maximum
of 8,000 characters."""
_StringType.__init__(self, collation)
sqltypes.CHAR.__init__(self, length, **kw)
+
class NCHAR(_StringType, sqltypes.NCHAR):
"""MSSQL NCHAR type.
_StringType.__init__(self, collation)
sqltypes.NCHAR.__init__(self, length, **kw)
+
class IMAGE(sqltypes.LargeBinary):
__visit_name__ = 'IMAGE'
+
class BIT(sqltypes.TypeEngine):
__visit_name__ = 'BIT'
class MONEY(sqltypes.TypeEngine):
__visit_name__ = 'MONEY'
+
class SMALLMONEY(sqltypes.TypeEngine):
__visit_name__ = 'SMALLMONEY'
+
class UNIQUEIDENTIFIER(sqltypes.TypeEngine):
__visit_name__ = "UNIQUEIDENTIFIER"
+
class SQL_VARIANT(sqltypes.TypeEngine):
__visit_name__ = 'SQL_VARIANT'
def visit_SQL_VARIANT(self, type_):
return 'SQL_VARIANT'
+
class MSExecutionContext(default.DefaultExecutionContext):
_enable_identity_insert = False
_select_lastrowid = False
else:
return engine.ResultProxy(self)
+
class MSSQLCompiler(compiler.SQLCompiler):
returning_precedes_values = True
fromhints=from_hints, **kw)
for t in [from_table] + extra_froms)
+
class MSSQLStrictCompiler(MSSQLCompiler):
"""A subclass of MSSQLCompiler which disables the usage of bind
parameters where not allowed natively by MS-SQL.
return super(MSSQLStrictCompiler, self).\
render_literal_value(value, type_)
+
class MSDDLCompiler(compiler.DDLCompiler):
def get_column_specification(self, column, **kwargs):
colspec = (self.preparer.format_column(column) + " "
result = '.'.join([self.quote(x, force) for x in schema.split('.')])
return result
+
def _db_plus_owner_listing(fn):
def wrap(dialect, connection, schema=None, **kw):
dbname, owner = _owner_plus_db(dialect, schema)
dbname, owner, schema, **kw)
return update_wrapper(wrap, fn)
+
def _db_plus_owner(fn):
def wrap(dialect, connection, tablename, schema=None, **kw):
dbname, owner = _owner_plus_db(dialect, schema)
tablename, dbname, owner, schema, **kw)
return update_wrapper(wrap, fn)
+
def _switch_db(dbname, connection, fn, *arg, **kw):
if dbname:
current_db = connection.scalar("select db_name()")
if dbname:
connection.execute("use %s" % current_db)
+
def _owner_plus_db(dialect, schema):
if not schema:
return None, dialect.default_schema_name
else:
return None, schema
+
class MSDialect(default.DefaultDialect):
name = 'mssql'
supports_default_values = True
ischema = MetaData()
+
class CoerceUnicode(TypeDecorator):
impl = Unicode
"""Include pyodbc's numeric processor.
"""
+
class _MSDate_mxodbc(_MSDate):
def bind_processor(self, dialect):
def process(value):
return None
return process
+
class _MSTime_mxodbc(_MSTime):
def bind_processor(self, dialect):
def process(value):
return None
return process
+
class MSExecutionContext_mxodbc(MSExecutionContext_pyodbc):
"""
The pyodbc execution context is useful for enabling
# is really only being used in cases where OUTPUT
# won't work.
+
class MSDialect_mxodbc(MxODBCConnector, MSDialect):
# this is only needed if "native ODBC" mode is used,
from ... import types as sqltypes, util, processors
import re
+
class _MSNumeric_pymssql(sqltypes.Numeric):
def result_processor(self, dialect, type_):
if not self.asdecimal:
else:
return sqltypes.Numeric.result_processor(self, dialect, type_)
+
class MSDialect_pymssql(MSDialect):
supports_sane_rowcount = False
driver = 'pymssql'
from ... import types as sqltypes, util
from ...util.compat import decimal
+
class _MSNumeric_pyodbc(sqltypes.Numeric):
"""Turns Decimals with adjusted() < 0 or > 7 into strings.
from .base import MSDialect, MSExecutionContext
from ... import engine
+
class MSExecutionContext_zxjdbc(MSExecutionContext):
_embedded_scope_identity = False
self.zerofill = zerofill
super(_NumericType, self).__init__(**kw)
+
class _FloatType(_NumericType, sqltypes.Float):
def __init__(self, precision=None, scale=None, asdecimal=True, **kw):
if isinstance(self, (REAL, DOUBLE)) and \
super(_FloatType, self).__init__(precision=precision, asdecimal=asdecimal, **kw)
self.scale = scale
+
class _IntegerType(_NumericType, sqltypes.Integer):
def __init__(self, display_width=None, **kw):
self.display_width = display_width
super(_IntegerType, self).__init__(**kw)
+
class _StringType(sqltypes.String):
"""Base for MySQL string types."""
super(DOUBLE, self).__init__(precision=precision, scale=scale,
asdecimal=asdecimal, **kw)
+
class REAL(_FloatType, sqltypes.REAL):
"""MySQL REAL type."""
super(REAL, self).__init__(precision=precision, scale=scale,
asdecimal=asdecimal, **kw)
+
class FLOAT(_FloatType, sqltypes.FLOAT):
"""MySQL FLOAT type."""
def bind_processor(self, dialect):
return None
+
class INTEGER(_IntegerType, sqltypes.INTEGER):
"""MySQL INTEGER type."""
"""
super(INTEGER, self).__init__(display_width=display_width, **kw)
+
class BIGINT(_IntegerType, sqltypes.BIGINT):
"""MySQL BIGINTEGER type."""
"""
super(BIGINT, self).__init__(display_width=display_width, **kw)
+
class MEDIUMINT(_IntegerType):
"""MySQL MEDIUMINTEGER type."""
"""
super(MEDIUMINT, self).__init__(display_width=display_width, **kw)
+
class TINYINT(_IntegerType):
"""MySQL TINYINT type."""
"""
super(TINYINT, self).__init__(display_width=display_width, **kw)
+
class SMALLINT(_IntegerType, sqltypes.SMALLINT):
"""MySQL SMALLINTEGER type."""
"""
super(SMALLINT, self).__init__(display_width=display_width, **kw)
+
class BIT(sqltypes.TypeEngine):
"""MySQL BIT type.
return value
return process
+
class TIME(sqltypes.TIME):
"""MySQL TIME type.
return None
return process
+
class TIMESTAMP(sqltypes.TIMESTAMP):
"""MySQL TIMESTAMP type."""
__visit_name__ = 'TIMESTAMP'
+
class YEAR(sqltypes.TypeEngine):
"""MySQL YEAR type, for single byte storage of years 1901-2155."""
def __init__(self, display_width=None):
self.display_width = display_width
+
class TEXT(_StringType, sqltypes.TEXT):
"""MySQL TEXT type, for text up to 2^16 characters."""
"""
super(TEXT, self).__init__(length=length, **kw)
+
class TINYTEXT(_StringType):
"""MySQL TINYTEXT type, for text up to 2^8 characters."""
"""
super(TINYTEXT, self).__init__(**kwargs)
+
class MEDIUMTEXT(_StringType):
"""MySQL MEDIUMTEXT type, for text up to 2^24 characters."""
"""
super(MEDIUMTEXT, self).__init__(**kwargs)
+
class LONGTEXT(_StringType):
"""MySQL LONGTEXT type, for text up to 2^32 characters."""
"""
super(VARCHAR, self).__init__(length=length, **kwargs)
+
class CHAR(_StringType, sqltypes.CHAR):
"""MySQL CHAR type, for fixed-length character data."""
"""
super(CHAR, self).__init__(length=length, **kwargs)
+
class NVARCHAR(_StringType, sqltypes.NVARCHAR):
"""MySQL NVARCHAR type.
super(NCHAR, self).__init__(length=length, **kwargs)
-
-
class TINYBLOB(sqltypes._Binary):
"""MySQL TINYBLOB type, for binary data up to 2^8 bytes."""
__visit_name__ = 'TINYBLOB'
+
class MEDIUMBLOB(sqltypes._Binary):
"""MySQL MEDIUMBLOB type, for binary data up to 2^24 bytes."""
__visit_name__ = 'MEDIUMBLOB'
+
class LONGBLOB(sqltypes._Binary):
"""MySQL LONGBLOB type, for binary data up to 2^32 bytes."""
__visit_name__ = 'LONGBLOB'
+
class ENUM(sqltypes.Enum, _StringType):
"""MySQL ENUM type."""
kw['strict'] = self.strict
return sqltypes.Enum.adapt(self, impltype, **kw)
+
class SET(_StringType):
"""MySQL SET type."""
'year': YEAR,
}
+
class MySQLExecutionContext(default.DefaultExecutionContext):
def should_autocommit_text(self, statement):
return AUTOCOMMIT_RE.match(statement)
+
class MySQLCompiler(compiler.SQLCompiler):
render_table_with_column_in_update_from = True
(self.preparer.format_table(constraint.table),
qual, const)
+
class MySQLTypeCompiler(compiler.GenericTypeCompiler):
def _extend_numeric(self, type_, spec):
"Extend a numeric-type declaration with MySQL specific extensions."
return tuple([self.quote_identifier(i) for i in ids if i is not None])
+
class MySQLDialect(default.DefaultDialect):
"""Details of the MySQL dialect. Not used directly in application code."""
rp.close()
return rows
+
class ReflectedState(object):
"""Stores raw information about a SHOW CREATE TABLE statement."""
self.keys = []
self.constraints = []
+
class MySQLTableDefinitionParser(object):
"""Parses the results of a SHOW CREATE TABLE statement."""
return (_re_compile(regex), cleanup)
+
def _re_compile(regex):
"""Compile a string to regex, I and UNICODE."""
from ... import util
+
class MySQLExecutionContext_mysqlconnector(MySQLExecutionContext):
def get_lastrowid(self):
def post_process_text(self, text):
return text.replace('%', '%%')
+
class MySQLIdentifierPreparer_mysqlconnector(MySQLIdentifierPreparer):
def _escape_identifier(self, value):
value = value.replace(self.escape_quote, self.escape_to_quote)
return value.replace("%", "%%")
+
class _myconnpyBIT(BIT):
def result_processor(self, dialect, coltype):
"""MySQL-connector already converts mysql bits, so."""
return None
+
class MySQLDialect_mysqlconnector(MySQLDialect):
driver = 'mysqlconnector'
supports_unicode_statements = True
MySQLDBConnector
)
+
class MySQLExecutionContext_mysqldb(MySQLDBExecutionContext, MySQLExecutionContext):
pass
class MySQLIdentifierPreparer_mysqldb(MySQLDBIdentifierPreparer, MySQLIdentifierPreparer):
pass
+
class MySQLDialect_mysqldb(MySQLDBConnector, MySQLDialect):
execution_ctx_cls = MySQLExecutionContext_mysqldb
statement_compiler = MySQLCompiler_mysqldb
def plain_query(self):
return self.execution_options.get('_oursql_plain_query', False)
+
class MySQLDialect_oursql(MySQLDialect):
driver = 'oursql'
# Py2K
from .mysqldb import MySQLDialect_mysqldb
+
class MySQLDialect_pymysql(MySQLDialect_mysqldb):
driver = 'pymysql'
from ... import util
import re
+
class MySQLExecutionContext_pyodbc(MySQLExecutionContext):
def get_lastrowid(self):
cursor.close()
return lastrowid
+
class MySQLDialect_pyodbc(PyODBCConnector, MySQLDialect):
supports_unicode_statements = False
execution_ctx_cls = MySQLExecutionContext_pyodbc
from ...connectors.zxJDBC import ZxJDBCConnector
from .base import BIT, MySQLDialect, MySQLExecutionContext
+
class _ZxJDBCBit(BIT):
def result_processor(self, dialect, coltype):
"""Converts boolean or byte arrays from MySQL Connector/J to longs."""
NO_ARG_FNS = set('UID CURRENT_DATE SYSDATE USER '
'CURRENT_TIME CURRENT_TIMESTAMP'.split())
+
class RAW(sqltypes._Binary):
__visit_name__ = 'RAW'
OracleRaw = RAW
+
class NCLOB(sqltypes.Text):
__visit_name__ = 'NCLOB'
+
class VARCHAR2(VARCHAR):
__visit_name__ = 'VARCHAR2'
NVARCHAR2 = NVARCHAR
+
class NUMBER(sqltypes.Numeric, sqltypes.Integer):
__visit_name__ = 'NUMBER'
super(DOUBLE_PRECISION, self).__init__(precision=precision, scale=scale, asdecimal=asdecimal)
+
class BFILE(sqltypes.LargeBinary):
__visit_name__ = 'BFILE'
+
class LONG(sqltypes.Text):
__visit_name__ = 'LONG'
+
class INTERVAL(sqltypes.TypeEngine):
__visit_name__ = 'INTERVAL'
def _type_affinity(self):
return sqltypes.Interval
+
class ROWID(sqltypes.TypeEngine):
"""Oracle ROWID type.
def visit_ROWID(self, type_):
return "ROWID"
+
class OracleCompiler(compiler.SQLCompiler):
"""Oracle compiler modifies the lexical structure of Select
statements to work under non-ANSI configured Oracle databases, if
else:
return super(OracleCompiler, self).for_update_clause(select)
+
class OracleDDLCompiler(compiler.DDLCompiler):
def define_constraint_cascades(self, constraint):
return super(OracleDDLCompiler, self).\
visit_create_index(create, include_schema=True)
+
class OracleIdentifierPreparer(compiler.IdentifierPreparer):
reserved_words = set([x.lower() for x in RESERVED_WORDS])
self.dialect.identifier_preparer.format_sequence(seq) +
".nextval FROM DUAL", type_)
+
class OracleDialect(default.DefaultDialect):
name = 'oracle'
supports_alter = True
from . import base as oracle
from ...engine import result as _result
from sqlalchemy import types as sqltypes, util, exc, processors
-from datetime import datetime
import random
import collections
from sqlalchemy.util.compat import decimal
import re
+
class _OracleNumeric(sqltypes.Numeric):
def bind_processor(self, dialect):
# cx_oracle accepts Decimal objects and floats
return super(_OracleNumeric, self).\
result_processor(dialect, coltype)
+
class _OracleDate(sqltypes.Date):
def bind_processor(self, dialect):
return None
return value
return process
+
class _LOBMixin(object):
def result_processor(self, dialect, coltype):
if not dialect.auto_convert_lobs:
return value
return process
+
class _NativeUnicodeMixin(object):
# Py3K
#pass
# unicode in all cases, so the "native_unicode" flag
# will be set for the default String.result_processor.
+
class _OracleChar(_NativeUnicodeMixin, sqltypes.CHAR):
def get_dbapi_type(self, dbapi):
return dbapi.FIXED_CHAR
+
class _OracleNVarChar(_NativeUnicodeMixin, sqltypes.NVARCHAR):
def get_dbapi_type(self, dbapi):
return getattr(dbapi, 'UNICODE', dbapi.STRING)
+
class _OracleText(_LOBMixin, sqltypes.Text):
def get_dbapi_type(self, dbapi):
return dbapi.CLOB
+
class _OracleString(_NativeUnicodeMixin, sqltypes.String):
pass
+
class _OracleUnicodeText(_LOBMixin, _NativeUnicodeMixin, sqltypes.UnicodeText):
def get_dbapi_type(self, dbapi):
return dbapi.NCLOB
return string_processor(lob_processor(value))
return process
+
class _OracleInteger(sqltypes.Integer):
def result_processor(self, dialect, coltype):
def to_int(val):
return val
return to_int
+
class _OracleBinary(_LOBMixin, sqltypes.LargeBinary):
def get_dbapi_type(self, dbapi):
return dbapi.BLOB
def bind_processor(self, dialect):
return None
+
class _OracleInterval(oracle.INTERVAL):
def get_dbapi_type(self, dbapi):
return dbapi.INTERVAL
+
class _OracleRaw(oracle.RAW):
pass
+
class _OracleRowid(oracle.ROWID):
def get_dbapi_type(self, dbapi):
return dbapi.ROWID
+
class OracleCompiler_cx_oracle(OracleCompiler):
def bindparam_string(self, name, quote=None, **kw):
if quote is True or quote is not False and \
return result
+
class OracleExecutionContext_cx_oracle_with_unicode(OracleExecutionContext_cx_oracle):
"""Support WITH_UNICODE in Python 2.xx.
return super(OracleExecutionContext_cx_oracle_with_unicode, self).\
_execute_scalar(unicode(stmt))
+
class ReturningResultProxy(_result.FullyBufferedResultProxy):
"""Result proxy which stuffs the _returning clause + outparams into the fetch."""
return collections.deque([tuple(self._returning_params["ret_%d" % i]
for i, c in enumerate(self._returning_params))])
+
class OracleDialect_cx_oracle(OracleDialect):
execution_ctx_cls = OracleExecutionContext_cx_oracle
statement_compiler = OracleCompiler_cx_oracle
SQLException = zxJDBC = None
+
class _ZxJDBCDate(sqltypes.Date):
def result_processor(self, dialect, coltype):
options = None
_existing_engine = None
+
def _log(option, opt_str, value, parser):
global logging
if not logging:
print "%20s\t%s" % (macro, file_config.get('db', macro))
sys.exit(0)
+
def _server_side_cursors(options, opt_str, value, parser):
db_opts['server_side_cursors'] = True
+
def _engine_strategy(options, opt_str, value, parser):
if value:
db_opts['strategy'] = value
pre_configure = []
post_configure = []
+
+
def pre(fn):
pre_configure.append(fn)
return fn
+
+
def post(fn):
post_configure.append(fn)
return fn
+
@pre
def _setup_options(opt, file_config):
global options
options = opt
+
@pre
def _monkeypatch_cdecimal(options, file_config):
if options.cdecimal:
- import sys
import cdecimal
sys.modules['decimal'] = cdecimal
+
@post
def _engine_uri(options, file_config):
global db_label, db_url
% db_label)
db_url = file_config.get('db', db_label)
+
@post
def _require(options, file_config):
if not(options.require or
continue
pkg_resources.require(requirement)
+
@post
def _engine_pool(options, file_config):
if options.mockpool:
from sqlalchemy import pool
db_opts['poolclass'] = pool.AssertionPool
+
@post
def _create_testing_engine(options, file_config):
from sqlalchemy.testing import engines, config
if options.mysql_engine:
table_options['mysql_engine'] = options.mysql_engine
+
@post
def _reverse_topological(options, file_config):
if options.reversetop:
topological.set = unitofwork.set = session.set = mapper.set = \
dependency.set = RandomSet
+
@post
def _requirements(options, file_config):
from sqlalchemy.testing import config
from sqlalchemy.testing import config
config.options = options
+
@post
def _setup_profiling(options, file_config):
from sqlalchemy.testing import profiling
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
