else:
return None
-registry = util.PluginLoader("sqlalchemy.dialects", auto_fn=_auto_fn)
\ No newline at end of file
+registry = util.PluginLoader("sqlalchemy.dialects", auto_fn=_auto_fn)
'TEXT', 'NUMERIC', 'FLOAT', 'TIMESTAMP', 'VARCHAR', 'CHAR', 'BLOB',
'dialect'
)
-
-
"""
-import datetime, re
+import datetime
from sqlalchemy import schema as sa_schema
from sqlalchemy import exc, types as sqltypes, sql, util
return self._parse_version_info(version)
-dialect = FBDialect_fdb
\ No newline at end of file
+dialect = FBDialect_fdb
from sqlalchemy.dialects.informix import base, informixdb
-base.dialect = informixdb.dialect
\ No newline at end of file
+base.dialect = informixdb.dialect
return process
colspecs = {
- sqltypes.DateTime : InfoDateTime,
+ sqltypes.DateTime: InfoDateTime,
sqltypes.TIMESTAMP: InfoDateTime,
sqltypes.Time: InfoTime,
}
class InformixDialect(default.DefaultDialect):
name = 'informix'
- max_identifier_length = 128 # adjusts at runtime based on server version
+ max_identifier_length = 128 # adjusts at runtime based on server version
type_compiler = InfoTypeCompiler
statement_compiler = InfoSQLCompiler
if coltype not in (0, 13) and default:
default = default.split()[-1]
- if coltype == 6: # Serial, mark as autoincrement
+ if coltype == 6: # Serial, mark as autoincrement
autoincrement = True
- if coltype == 0 or coltype == 13: # char, varchar
+ if coltype == 0 or coltype == 13: # char, varchar
coltype = ischema_names[coltype](collength)
if default:
default = "'%s'" % default
- elif coltype == 5: # decimal
+ elif coltype == 5: # decimal
precision, scale = (collength & 0xFF00) >> 8, collength & 0xFF
if scale == 255:
scale = 0
def fkey_rec():
return {
- 'name' : None,
- 'constrained_columns' : [],
- 'referred_schema' : None,
- 'referred_table' : None,
- 'referred_columns' : []
+ 'name': None,
+ 'constrained_columns': [],
+ 'referred_schema': None,
+ 'referred_table': None,
+ 'referred_columns': []
}
fkeys = util.defaultdict(fkey_rec)
'DATETIME2', 'DATETIMEOFFSET', 'DATE', 'TIME', 'SMALLDATETIME',
'BINARY', 'VARBINARY', 'BIT', 'REAL', 'IMAGE', 'TIMESTAMP',
'MONEY', 'SMALLMONEY', 'UNIQUEIDENTIFIER', 'SQL_VARIANT', 'dialect'
-)
\ No newline at end of file
+)
connectors = ["Provider=SQLOLEDB"]
if 'port' in keys:
- connectors.append ("Data Source=%s, %s" %
+ connectors.append("Data Source=%s, %s" %
(keys.get("host"), keys.get("port")))
else:
- connectors.append ("Data Source=%s" % keys.get("host"))
- connectors.append ("Initial Catalog=%s" % keys.get("database"))
+ connectors.append("Data Source=%s" % keys.get("host"))
+ connectors.append("Initial Catalog=%s" % keys.get("database"))
user = keys.get("user")
if user:
connectors.append("User Id=%s" % user)
connectors.append("Password=%s" % keys.get("password", ""))
else:
connectors.append("Integrated Security=SSPI")
- return [[";".join (connectors)], {}]
+ return [[";".join(connectors)], {}]
def is_disconnect(self, e, connection, cursor):
return isinstance(e, self.dbapi.adodbapi.DatabaseError) and \
return process
_reg = re.compile(r"(\d+)-(\d+)-(\d+)")
+
def result_processor(self, dialect, coltype):
def process(value):
if isinstance(value, datetime.datetime):
return process
_reg = re.compile(r"(\d+):(\d+):(\d+)(?:\.(\d{0,6}))?")
+
def result_processor(self, dialect, coltype):
def process(value):
if isinstance(value, datetime.datetime):
MSVariant = SQL_VARIANT
ischema_names = {
- 'int' : INTEGER,
+ 'int': INTEGER,
'bigint': BIGINT,
- 'smallint' : SMALLINT,
- 'tinyint' : TINYINT,
- 'varchar' : VARCHAR,
- 'nvarchar' : NVARCHAR,
- 'char' : CHAR,
- 'nchar' : NCHAR,
- 'text' : TEXT,
- 'ntext' : NTEXT,
- 'decimal' : DECIMAL,
- 'numeric' : NUMERIC,
- 'float' : FLOAT,
- 'datetime' : DATETIME,
- 'datetime2' : DATETIME2,
- 'datetimeoffset' : DATETIMEOFFSET,
+ 'smallint': SMALLINT,
+ 'tinyint': TINYINT,
+ 'varchar': VARCHAR,
+ 'nvarchar': NVARCHAR,
+ 'char': CHAR,
+ 'nchar': NCHAR,
+ 'text': TEXT,
+ 'ntext': NTEXT,
+ 'decimal': DECIMAL,
+ 'numeric': NUMERIC,
+ 'float': FLOAT,
+ 'datetime': DATETIME,
+ 'datetime2': DATETIME2,
+ 'datetimeoffset': DATETIMEOFFSET,
'date': DATE,
'time': TIME,
- 'smalldatetime' : SMALLDATETIME,
- 'binary' : BINARY,
- 'varbinary' : VARBINARY,
+ 'smalldatetime': SMALLDATETIME,
+ 'binary': BINARY,
+ 'varbinary': VARBINARY,
'bit': BIT,
- 'real' : REAL,
- 'image' : IMAGE,
+ 'real': REAL,
+ 'image': IMAGE,
'timestamp': TIMESTAMP,
'money': MONEY,
'smallmoney': SMALLMONEY,
return self._extend("TEXT", type_)
def visit_VARCHAR(self, type_):
- return self._extend("VARCHAR", type_,
- length = type_.length or 'max')
+ return self._extend("VARCHAR", type_, length=type_.length or 'max')
def visit_CHAR(self, type_):
return self._extend("CHAR", type_)
return self._extend("NCHAR", type_)
def visit_NVARCHAR(self, type_):
- return self._extend("NVARCHAR", type_,
- length = type_.length or 'max')
+ return self._extend("NVARCHAR", type_, length=type_.length or 'max')
def visit_date(self, type_):
if self.dialect.server_version_info < MS_2008_VERSION:
schema_name = "dbo"
colspecs = {
- sqltypes.DateTime : _MSDateTime,
- sqltypes.Date : _MSDate,
- sqltypes.Time : TIME,
+ sqltypes.DateTime: _MSDateTime,
+ sqltypes.Date: _MSDate,
+ sqltypes.Time: TIME,
}
ischema_names = ischema_names
"behaviors may not function properly. If using ODBC "
"with FreeTDS, ensure server version 7.0 or 8.0, not 4.2, "
"is configured in the FreeTDS configuration." %
- ".".join(str(x) for x in self.server_version_info) )
+ ".".join(str(x) for x in self.server_version_info))
if self.server_version_info >= MS_2005_VERSION and \
'implicit_returning' not in self.__dict__:
self.implicit_returning = True
sql.bindparam('schname', owner,
sqltypes.String(convert_unicode=True))
],
- typemap = {
+ typemap={
'name': sqltypes.Unicode()
}
)
sql.bindparam('schname', owner,
sqltypes.String(convert_unicode=True))
],
- typemap = {
- 'name': sqltypes.Unicode()
- }
+ typemap={'name': sqltypes.Unicode()}
),
)
for row in rp:
coltype = coltype(**kwargs)
cdict = {
- 'name' : name,
- 'type' : coltype,
- 'nullable' : nullable,
- 'default' : default,
+ 'name': name,
+ 'type': coltype,
+ 'nullable': nullable,
+ 'default': default,
'autoincrement': False,
}
cols.append(cdict)
RR.c.unique_constraint_name,
C.c.ordinal_position == R.c.ordinal_position
),
- order_by= [
- RR.c.constraint_name,
- R.c.ordinal_position])
+ order_by=[RR.c.constraint_name, R.c.ordinal_position]
+ )
# group rows by constraint ID, to handle multi-column FKs
fkeys = []
remote_cols.append(rcol)
return fkeys.values()
-
Column("CHECK_OPTION", String, key="check_option"),
Column("IS_UPDATABLE", String, key="is_updatable"),
schema="INFORMATION_SCHEMA")
-
_need_decimal_fix = True
colspecs = {
- sqltypes.Numeric : _MSNumeric_mxodbc,
- sqltypes.DateTime : _MSDateTime,
- sqltypes.Date : _MSDate_mxodbc,
- sqltypes.Time : _MSTime_mxodbc,
+ sqltypes.Numeric: _MSNumeric_mxodbc,
+ sqltypes.DateTime: _MSDateTime,
+ sqltypes.Date: _MSDate_mxodbc,
+ sqltypes.Time: _MSTime_mxodbc,
}
def __init__(self, description_encoding=None, **params):
self.description_encoding = description_encoding
dialect = MSDialect_mxodbc
-
sqltypes.Float: sqltypes.Float,
}
)
+
@classmethod
def dbapi(cls):
module = __import__('pymssql')
'union', 'unique', 'unlock', 'unsigned', 'update', 'usage', 'use',
'using', 'utc_date', 'utc_time', 'utc_timestamp', 'values', 'varbinary',
'varchar', 'varcharacter', 'varying', 'when', 'where', 'while', 'with',
- 'write', 'x509', 'xor', 'year_month', 'zerofill', # 5.0
- 'columns', 'fields', 'privileges', 'soname', 'tables', # 4.1
+ 'write', 'x509', 'xor', 'year_month', 'zerofill', # 5.0
+ 'columns', 'fields', 'privileges', 'soname', 'tables', # 4.1
'accessible', 'linear', 'master_ssl_verify_server_cert', 'range',
- 'read_only', 'read_write', # 5.1
+ 'read_only', 'read_write', # 5.1
])
AUTOCOMMIT_RE = re.compile(
def result_processor(self, dialect, coltype):
time = datetime.time
+
def process(value):
# convert from a timedelta value
if value is not None:
def bind_processor(self, dialect):
super_convert = super(ENUM, self).bind_processor(dialect)
+
def process(value):
if self.strict and value is not None and value not in self.enums:
raise exc.InvalidRequestError('"%s" not a valid value for '
def bind_processor(self, dialect):
super_convert = super(SET, self).bind_processor(dialect)
+
def process(value):
if value is None or isinstance(value, (int, long, basestring)):
pass
"""Overridden from base SQLCompiler value"""
extract_map = compiler.SQLCompiler.extract_map.copy()
- extract_map.update ({
- 'milliseconds': 'millisecond',
- })
+ extract_map.update({'milliseconds': 'millisecond'})
def visit_random_func(self, fn, **kw):
return "rand%s" % self.function_argspec(fn)
else:
return self._extend_numeric(type_,
"NUMERIC(%(precision)s, %(scale)s)" %
- {'precision': type_.precision, 'scale' : type_.scale})
+ {'precision': type_.precision, 'scale': type_.scale})
def visit_DECIMAL(self, type_):
if type_.precision is None:
else:
return self._extend_numeric(type_,
"DECIMAL(%(precision)s, %(scale)s)" %
- {'precision': type_.precision, 'scale' : type_.scale})
+ {'precision': type_.precision, 'scale': type_.scale})
def visit_DOUBLE(self, type_):
if type_.precision is not None and type_.scale is not None:
return self._extend_numeric(type_, "DOUBLE(%(precision)s, %(scale)s)" %
{'precision': type_.precision,
- 'scale' : type_.scale})
+ 'scale': type_.scale})
else:
return self._extend_numeric(type_, 'DOUBLE')
if type_.precision is not None and type_.scale is not None:
return self._extend_numeric(type_, "REAL(%(precision)s, %(scale)s)" %
{'precision': type_.precision,
- 'scale' : type_.scale})
+ 'scale': type_.scale})
else:
return self._extend_numeric(type_, 'REAL')
con_kw[opt] = spec[opt]
fkey_d = {
- 'name' : spec['name'],
- 'constrained_columns' : loc_names,
- 'referred_schema' : ref_schema,
- 'referred_table' : ref_name,
- 'referred_columns' : ref_names,
- 'options' : con_kw
+ 'name': spec['name'],
+ 'constrained_columns': loc_names,
+ 'referred_schema': ref_schema,
+ 'referred_table': ref_name,
+ 'referred_columns': ref_names,
+ 'options': con_kw
}
fkeys.append(fkey_d)
return fkeys
"""Compile a string to regex, I and UNICODE."""
return re.compile(regex, re.I | re.UNICODE)
-
if code:
return int(code)
-dialect = MySQLDialect_gaerdbms
\ No newline at end of file
+dialect = MySQLDialect_gaerdbms
# Q: why didn't we need all these "plain_query" overrides earlier ?
# am i on a newer/older version of OurSQL ?
def has_table(self, connection, table_name, schema=None):
- return MySQLDialect.has_table(self,
- connection.connect().\
- execution_options(_oursql_plain_query=True),
- table_name, schema)
+ return MySQLDialect.has_table(
+ self,
+ connection.connect().execution_options(_oursql_plain_query=True),
+ table_name,
+ schema
+ )
def get_table_options(self, connection, table_name, schema=None, **kw):
- return MySQLDialect.get_table_options(self,
- connection.connect().\
- execution_options(_oursql_plain_query=True),
- table_name,
- schema = schema,
- **kw
+ return MySQLDialect.get_table_options(
+ self,
+ connection.connect().execution_options(_oursql_plain_query=True),
+ table_name,
+ schema=schema,
+ **kw
)
def get_columns(self, connection, table_name, schema=None, **kw):
- return MySQLDialect.get_columns(self,
- connection.connect().\
- execution_options(_oursql_plain_query=True),
- table_name,
- schema=schema,
- **kw
+ return MySQLDialect.get_columns(
+ self,
+ connection.connect().execution_options(_oursql_plain_query=True),
+ table_name,
+ schema=schema,
+ **kw
)
def get_view_names(self, connection, schema=None, **kw):
- return MySQLDialect.get_view_names(self,
- connection.connect().\
- execution_options(_oursql_plain_query=True),
- schema=schema,
- **kw
+ return MySQLDialect.get_view_names(
+ self,
+ connection.connect().execution_options(_oursql_plain_query=True),
+ schema=schema,
+ **kw
)
def get_table_names(self, connection, schema=None, **kw):
- return MySQLDialect.get_table_names(self,
- connection.connect().\
- execution_options(_oursql_plain_query=True),
- schema
+ return MySQLDialect.get_table_names(
+ self,
+ connection.connect().execution_options(_oursql_plain_query=True),
+ schema
)
def get_schema_names(self, connection, **kw):
- return MySQLDialect.get_schema_names(self,
- connection.connect().\
- execution_options(_oursql_plain_query=True),
- **kw
+ return MySQLDialect.get_schema_names(
+ self,
+ connection.connect().execution_options(_oursql_plain_query=True),
+ **kw
)
def initialize(self, connection):
return MySQLDialect.initialize(
- self,
- connection.execution_options(_oursql_plain_query=True)
- )
+ self,
+ connection.execution_options(_oursql_plain_query=True)
+ )
def _show_create_table(self, connection, table, charset=None,
full_name=None):
- return MySQLDialect._show_create_table(self,
- connection.contextual_connect(close_with_result=True).
- execution_options(_oursql_plain_query=True),
- table, charset, full_name)
+ return MySQLDialect._show_create_table(
+ self,
+ connection.contextual_connect(close_with_result=True).
+ execution_options(_oursql_plain_query=True),
+ table, charset, full_name
+ )
def is_disconnect(self, e, connection, cursor):
if isinstance(e, self.dbapi.ProgrammingError):
driver = 'pymysql'
description_encoding = None
+
@classmethod
def dbapi(cls):
return __import__('pymysql')
-dialect = MySQLDialect_pymysql
\ No newline at end of file
+dialect = MySQLDialect_pymysql
"""
-import random, re
+import re
-from sqlalchemy import schema as sa_schema
-from sqlalchemy import util, sql, log
+from sqlalchemy import util, sql
from sqlalchemy.engine import default, base, reflection
from sqlalchemy.sql import compiler, visitors, expression
from sqlalchemy.sql import operators as sql_operators, functions as sql_functions
class DOUBLE_PRECISION(sqltypes.Numeric):
__visit_name__ = 'DOUBLE_PRECISION'
+
def __init__(self, precision=None, scale=None, asdecimal=None):
if asdecimal is None:
asdecimal = False
return dbapi.NUMBER
colspecs = {
- sqltypes.Boolean : _OracleBoolean,
- sqltypes.Interval : INTERVAL,
+ sqltypes.Boolean: _OracleBoolean,
+ sqltypes.Interval: INTERVAL,
}
ischema_names = {
- 'VARCHAR2' : VARCHAR,
- 'NVARCHAR2' : NVARCHAR,
- 'CHAR' : CHAR,
- 'DATE' : DATE,
- 'NUMBER' : NUMBER,
- 'BLOB' : BLOB,
- 'BFILE' : BFILE,
- 'CLOB' : CLOB,
- 'NCLOB' : NCLOB,
- 'TIMESTAMP' : TIMESTAMP,
- 'TIMESTAMP WITH TIME ZONE' : TIMESTAMP,
- 'INTERVAL DAY TO SECOND' : INTERVAL,
- 'RAW' : RAW,
- 'FLOAT' : FLOAT,
- 'DOUBLE PRECISION' : DOUBLE_PRECISION,
- 'LONG' : LONG,
+ 'VARCHAR2': VARCHAR,
+ 'NVARCHAR2': NVARCHAR,
+ 'CHAR': CHAR,
+ 'DATE': DATE,
+ 'NUMBER': NUMBER,
+ 'BLOB': BLOB,
+ 'BFILE': BFILE,
+ 'CLOB': CLOB,
+ 'NCLOB': NCLOB,
+ 'TIMESTAMP': TIMESTAMP,
+ 'TIMESTAMP WITH TIME ZONE': TIMESTAMP,
+ 'INTERVAL DAY TO SECOND': INTERVAL,
+ 'RAW': RAW,
+ 'FLOAT': FLOAT,
+ 'DOUBLE PRECISION': DOUBLE_PRECISION,
+ 'LONG': LONG,
}
def visit_RAW(self, type_):
if type_.length:
- return "RAW(%(length)s)" % {'length' : type_.length}
+ return "RAW(%(length)s)" % {'length': type_.length}
else:
return "RAW"
(colname, orig_colname, coltype, length, precision, scale, nullable, default) = \
(self.normalize_name(row[0]), row[0], row[1], row[2], row[3], row[4], row[5] == 'Y', row[6])
- if coltype == 'NUMBER' :
+ if coltype == 'NUMBER':
coltype = NUMBER(precision, scale)
elif coltype in ('VARCHAR2', 'NVARCHAR2', 'CHAR'):
coltype = self.ischema_names.get(coltype)(length)
"""
- requested_schema = schema # to check later on
+ requested_schema = schema # to check later on
resolve_synonyms = kw.get('oracle_resolve_synonyms', False)
dblink = kw.get('dblink', '')
info_cache = kw.get('info_cache')
def fkey_rec():
return {
- 'name' : None,
- 'constrained_columns' : [],
- 'referred_schema' : None,
- 'referred_table' : None,
- 'referred_columns' : []
+ 'name': None,
+ 'constrained_columns': [],
+ 'referred_schema': None,
+ 'referred_table': None,
+ 'referred_columns': []
}
fkeys = util.defaultdict(fkey_rec)
fstring = "%.10f"
else:
fstring = "%%.%df" % self.scale
+
def to_decimal(value):
if value is None:
return None
return value
else:
return decimal.Decimal(fstring % value)
+
return to_decimal
else:
if self.precision is None and self.scale is None:
colspecs = colspecs = {
sqltypes.Numeric: _OracleNumeric,
- sqltypes.Date : _OracleDate, # generic type, assume datetime.date is desired
+ sqltypes.Date: _OracleDate, # generic type, assume datetime.date is desired
oracle.DATE: oracle.DATE, # non generic type - passthru
- sqltypes.LargeBinary : _OracleBinary,
- sqltypes.Boolean : oracle._OracleBoolean,
- sqltypes.Interval : _OracleInterval,
- oracle.INTERVAL : _OracleInterval,
- sqltypes.Text : _OracleText,
- sqltypes.String : _OracleString,
- sqltypes.UnicodeText : _OracleUnicodeText,
- sqltypes.CHAR : _OracleChar,
- sqltypes.Integer : _OracleInteger, # this is only needed for OUT parameters.
- # it would be nice if we could not use it otherwise.
+ sqltypes.LargeBinary: _OracleBinary,
+ sqltypes.Boolean: oracle._OracleBoolean,
+ sqltypes.Interval: _OracleInterval,
+ oracle.INTERVAL: _OracleInterval,
+ sqltypes.Text: _OracleText,
+ sqltypes.String: _OracleString,
+ sqltypes.UnicodeText: _OracleUnicodeText,
+ sqltypes.CHAR: _OracleChar,
+
+ # this is only needed for OUT parameters.
+ # it would be nice if we could not use it otherwise.
+ sqltypes.Integer: _OracleInteger,
+
oracle.RAW: _OracleRaw,
sqltypes.Unicode: _OracleNVarChar,
- sqltypes.NVARCHAR : _OracleNVarChar,
+ sqltypes.NVARCHAR: _OracleNVarChar,
oracle.ROWID: _OracleRowid,
}
self.dbapi.BLOB: oracle.BLOB(),
self.dbapi.BINARY: oracle.RAW(),
}
+
@classmethod
def dbapi(cls):
import cx_Oracle
return
cx_Oracle = self.dbapi
+
def output_type_handler(cursor, name, defaultType,
size, precision, scale):
# convert all NUMBER with precision + positive scale to Decimal
self.binds[bindparam.key] = bindparam
binds.append(self.bindparam_string(self._truncate_bindparam(bindparam)))
- return 'RETURNING ' + ', '.join(columns) + " INTO " + ", ".join(binds)
+ return 'RETURNING ' + ', '.join(columns) + " INTO " + ", ".join(binds)
class OracleExecutionContext_zxjdbc(OracleExecutionContext):
colspecs = util.update_copy(
OracleDialect.colspecs,
{
- sqltypes.Date : _ZxJDBCDate,
+ sqltypes.Date: _ZxJDBCDate,
sqltypes.Numeric: _ZxJDBCNumeric
}
)
from java.sql import SQLException
from com.ziclix.python.sql import zxJDBC
from com.ziclix.python.sql.handler import OracleDataHandler
- class OracleReturningDataHandler(OracleDataHandler):
+ class OracleReturningDataHandler(OracleDataHandler):
"""zxJDBC DataHandler that specially handles ReturningParam."""
def setJDBCObject(self, statement, index, object, dbtype=None):
if type(object) is ReturningParam:
statement.registerReturnParameter(index, object.type)
elif dbtype is None:
- OracleDataHandler.setJDBCObject(self, statement, index, object)
+ OracleDataHandler.setJDBCObject(
+ self, statement, index, object)
else:
- OracleDataHandler.setJDBCObject(self, statement, index, object, dbtype)
+ OracleDataHandler.setJDBCObject(
+ self, statement, index, object, dbtype)
self.DataHandler = OracleReturningDataHandler
def initialize(self, connection):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
