rattail/sqlbase7-sa
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merged sqlalchemy-0.5 branch.

Browse source
This commit is contained in:
Lance Edgar 2010-05-21 20:06:47 -05:00
commit 21214df490
7 changed files with 224 additions and 135 deletions
Download patch file Download diff file Expand all files Collapse all files

3
setup.py
View file

@ -40,9 +40,10 @@ setup(
packages = find_packages(),
install_requires = [
'SQLAlchemy>=0.6.0',
'SQLAlchemy>=0.5,<=0.6.99',
],
# This is only used by SQLAlchemy 0.6.
entry_points = {
'sqlalchemy.dialects' : [
'sqlbase7 = sqlbase7_sa:base.dialect',

25
sqlbase7_sa/__init__.py
View file

@ -23,11 +23,26 @@
################################################################################
from sqlbase7_sa import base
from sqlbase7_sa import pyodbc
from sqlbase7_sa._version import __version__
import sqlalchemy
from pkg_resources import parse_version
# default dialect
base.dialect = pyodbc.dialect
if parse_version(sqlalchemy.__version__) <= parse_version('0.5.99'):
# SQLAlchemy 0.5 doesn't support user-contributed dialects "directly"
# by way of setuptools entry points, so we must monkey-patch it in
# order to add ours.
import sqlalchemy.databases, sqlbase7_sa, sys
sqlalchemy.databases.sqlbase7 = sqlbase7_sa
sys.modules['sqlalchemy.databases.sqlbase7'] = sqlbase7_sa
# SQLAlchemy will be expecting us to have a 'dialect' attribute.
import sqlbase7_sa.sqlbase7_sa05
dialect = sqlbase7_sa.sqlbase7_sa05.SQLBase7Dialect_SA05
else:
# SQLAlchemy 0.6 is much nicer to play with.
import sqlbase7_sa.sqlbase7_sa06
base = sqlbase7_sa.sqlbase7_sa06

44
sqlbase7_sa/pyodbc.py
View file

@ -1,44 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
################################################################################
#
# SQLBase7-SA -- SQLAlchemy driver/dialect for Centura SQLBase v7
# Copyright © 2010 Lance Edgar
#
# This file is part of SQLBase7-SA.
#
# SQLBase7-SA is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# SQLBase7-SA is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with SQLBase7-SA. If not, see <http://www.gnu.org/licenses/>.
#
################################################################################
from sqlalchemy.connectors.pyodbc import PyODBCConnector
from sqlbase7_sa.base import SQLBase7Dialect
class SQLBase7_pyodbc(PyODBCConnector, SQLBase7Dialect):
def create_connect_args(self, url):
connection_string = ';'.join((
"DRIVER={Centura SQLBase 3.5 32-bit Driver -NT & Win95}",
"SERVER=%s" % url.host,
"DATABASE=%s" % url.database,
"UID=%s" % url.username,
"PWD=%s" % url.password,
))
return [connection_string], {}
dialect = SQLBase7_pyodbc

116
sqlbase7_sa/base.py → sqlbase7_sa/sqlbase7.py
View file

@ -25,10 +25,17 @@
from sqlalchemy.engine.default import DefaultDialect
from sqlalchemy import types, and_
from sqlalchemy.sql.compiler import SQLCompiler
from sqlalchemy.sql.expression import Join
import sqlalchemy
from pkg_resources import parse_version
if parse_version(sqlalchemy.__version__) <= parse_version('0.5.99'):
from sqlalchemy.sql.compiler import DefaultCompiler as CompilerBase
else:
from sqlalchemy.sql.compiler import SQLCompiler as CompilerBase
class LimitClauseNotSupported(Exception):
def __init__(self, limit, offset):
@ -39,7 +46,7 @@ class LimitClauseNotSupported(Exception):
return "Centura SQLBase 7.5.1 doesn't support a LIMIT clause for the SELECT statement (received: limit = %u, offset = %u)" % (self.limit, self.offset)
class SQLBase7Compiler(SQLCompiler):
class SQLBase7Compiler(CompilerBase):
# Most of the code below was copied from the Oracle dialect. Thanks to Michael Bayer
# for pointing that out. Oh, and for writing SQLAlchemy; that was pretty cool.
@ -58,7 +65,7 @@ class SQLBase7Compiler(SQLCompiler):
raise LimitClauseNotSupported(select._limit, select._offset)
kwargs['iswrapper'] = getattr(select, '_is_wrapper', False)
return SQLCompiler.visit_select(self, select, **kwargs)
return super(SQLBase7Compiler, self).visit_select(select, **kwargs)
def _get_join_whereclause(self, froms):
clauses = []
@ -77,27 +84,16 @@ class SQLBase7Compiler(SQLCompiler):
return and_(*clauses)
return None
def visit_ilike_op(self, binary, **kw):
escape = binary.modifiers.get("escape", None)
return '@lower(%s) LIKE @lower(%s)' % (
self.process(binary.left, **kw),
self.process(binary.right, **kw)) \
+ (escape and ' ESCAPE \'%s\'' % escape or '')
class SQLBase7Dialect(DefaultDialect):
name = 'sqlbase7'
max_identifier_length = 18
# # Hmm, it'd be great if these actually did something...
# supports_unicode_statements = False
# supports_unicode_binds = False
statement_compiler = SQLBase7Compiler
type_map = {
max_identifier_length = 18
_type_map = {
'CHAR' : types.CHAR,
'DATE' : types.DATE,
'DECIMAL' : types.DECIMAL,
@ -108,75 +104,29 @@ class SQLBase7Dialect(DefaultDialect):
'VARCHAR' : types.VARCHAR,
}
def _check_unicode_returns(self, connection):
return False
def get_table_names(self, connection, schema=None, **kw):
if schema is None:
schema = ''
else:
schema = '%s.' % schema
cursor = connection.connection.cursor()
table_names = [row.NAME for row in cursor.execute(
"SELECT NAME FROM %sSYSTABLES WHERE REMARKS IS NOT NULL" % schema
)]
cursor.close()
return table_names
def get_columns(self, connection, table_name, schema=None, **kw):
if schema is None:
schema = ''
else:
schema = '%s.' % schema
cursor = connection.connection.cursor()
columns = []
for row in cursor.execute("SELECT NAME,COLTYPE,NULLS FROM %sSYSCOLUMNS WHERE TBNAME = '%s'" % (schema, table_name)):
columns.append({
'name' : row.NAME,
'type' : self.type_map[row.COLTYPE],
'nullable' : row.NULLS == 'Y',
'default' : None,
'autoincrement' : False,
})
cursor.close()
return columns
def get_primary_keys(self, connection, table_name, schema=None, **kw):
if schema is None:
schema = ''
else:
schema = '%s.' % schema
cursor = connection.connection.cursor()
primary_keys = [row.COLNAME for row in cursor.execute(
"SELECT COLNAME FROM %sSYSPKCONSTRAINTS WHERE NAME = '%s' ORDER BY PKCOLSEQNUM" % (schema, table_name)
)]
cursor.close()
return primary_keys
def get_foreign_keys(self, connection, table_name, schema=None, **kw):
return []
def get_indexes(self, connection, table_name, schema=None, **kw):
return []
def create_connect_args(self, url):
connection_string = ';'.join((
"DRIVER={Centura SQLBase 3.5 32-bit Driver -NT & Win95}",
"SERVER=%s" % url.host,
"DATABASE=%s" % url.database,
"UID=%s" % url.username,
"PWD=%s" % url.password,
))
return [connection_string], {}
def get_default_schema_name(self, connection):
return 'SYSADM'
def do_execute(self, cursor, statement, parameters, context=None):
# For some (perhaps good?) reason, the SQLBase ODBC driver doesn't like
# parameters if they're of Unicode or Long type. I'd hoped at first that
# the "supports_unicode_binds" attribute would take care of the Unicode
# problem but it didn't seem to. And now that the Long parameters seem
# to throw the same error, so...
_parameters = []
for parameter in parameters:
# problem but it didn't seem to. And now the Long parameters seem to
# throw the same error, so...
parameters = list(parameters)
for i, parameter in enumerate(parameters):
if isinstance(parameter, unicode):
parameter = str(parameter)
parameters[i] = str(parameter)
elif isinstance(parameter, long):
parameter = int(parameter)
_parameters.append(parameter)
parameters = tuple(_parameters)
cursor.execute(statement, parameters)
parameters[i] = int(parameter)
super(SQLBase7Dialect, self).do_execute(cursor, statement, tuple(parameters), context)

69
sqlbase7_sa/sqlbase7_sa05.py Normal file
View file

@ -0,0 +1,69 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
################################################################################
#
# SQLBase7-SA -- SQLAlchemy driver/dialect for Centura SQLBase v7
# Copyright © 2010 Lance Edgar
#
# This file is part of SQLBase7-SA.
#
# SQLBase7-SA is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# SQLBase7-SA is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with SQLBase7-SA. If not, see <http://www.gnu.org/licenses/>.
#
################################################################################
from sqlalchemy import types, Column, PrimaryKeyConstraint
from sqlalchemy.sql.compiler import OPERATORS
from sqlalchemy.sql import operators
from sqlbase7_sa.sqlbase7 import SQLBase7Dialect
OPERATORS[operators.ilike_op] = lambda x, y, escape=None: "@lower(%s) LIKE @lower(%s)" % (x, y) + (escape and ' ESCAPE \'%s\'' % escape or '')
class SQLBase7Dialect_SA05(SQLBase7Dialect):
@classmethod
def dbapi(cls):
import pyodbc
return pyodbc
def table_names(self, connection, schema):
cursor = connection.connection.cursor()
table_names = [row.NAME for row in cursor.execute(
"SELECT NAME FROM %s.SYSTABLES WHERE REMARKS IS NOT NULL" % schema
)]
cursor.close()
return table_names
def reflecttable(self, connection, table, include_columns=None):
if table.schema is None:
table.schema = connection.default_schema_name()
sql = "SELECT NAME,COLTYPE,NULLS FROM %s.SYSCOLUMNS WHERE TBNAME = '%s'" % (table.schema, table.name)
if include_columns:
sql += " AND NAME NOT IN (%s)" % ','.join(include_columns)
cursor = connection.connection.cursor()
for row in cursor.execute(sql):
table.append_column(Column(row.NAME, self._type_map[row.COLTYPE]))
cursor.close()
cursor = connection.connection.cursor()
key_columns = [row.COLNAME for row in cursor.execute(
"SELECT COLNAME FROM %s.SYSPKCONSTRAINTS WHERE NAME = '%s' ORDER BY PKCOLSEQNUM" % (table.schema, table.name)
)]
if key_columns:
table.append_constraint(PrimaryKeyConstraint(*key_columns))
cursor.close()

99
sqlbase7_sa/sqlbase7_sa06.py Normal file
View file

@ -0,0 +1,99 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
################################################################################
#
# SQLBase7-SA -- SQLAlchemy driver/dialect for Centura SQLBase v7
# Copyright © 2010 Lance Edgar
#
# This file is part of SQLBase7-SA.
#
# SQLBase7-SA is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# SQLBase7-SA is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with SQLBase7-SA. If not, see <http://www.gnu.org/licenses/>.
#
################################################################################
from sqlbase7_sa.sqlbase7 import SQLBase7Compiler, SQLBase7Dialect
from sqlalchemy.connectors.pyodbc import PyODBCConnector
class SQLBase7Compiler_SA06(SQLBase7Compiler):
def visit_ilike_op(self, binary, **kw):
escape = binary.modifiers.get("escape", None)
return '@lower(%s) LIKE @lower(%s)' % (
self.process(binary.left, **kw),
self.process(binary.right, **kw)) \
+ (escape and ' ESCAPE \'%s\'' % escape or '')
class SQLBase7Dialect_SA06(SQLBase7Dialect):
statement_compiler = SQLBase7Compiler_SA06
def _get_default_schema_name(self, connection):
return 'SYSADM'
def _check_unicode_returns(self, connection):
return False
def get_table_names(self, connection, schema=None, **kw):
cursor = connection.connection.cursor()
table_names = [row.NAME for row in cursor.execute(
"SELECT NAME FROM %s.SYSTABLES WHERE REMARKS IS NOT NULL" % schema
)]
cursor.close()
return table_names
def get_columns(self, connection, table_name, schema=None, **kw):
if schema is None:
schema = self.get_default_schema_name(connection)
cursor = connection.connection.cursor()
columns = []
for row in cursor.execute("SELECT NAME,COLTYPE,NULLS FROM %s.SYSCOLUMNS WHERE TBNAME = '%s'" % (schema, table_name)):
columns.append({
'name' : row.NAME,
'type' : self._type_map[row.COLTYPE],
'nullable' : row.NULLS == 'Y',
'default' : None,
'autoincrement' : False,
})
cursor.close()
return columns
def get_primary_keys(self, connection, table_name, schema=None, **kw):
if schema is None:
schema = self.get_default_schema_name(connection)
cursor = connection.connection.cursor()
primary_keys = [row.COLNAME for row in cursor.execute(
"SELECT COLNAME FROM %s.SYSPKCONSTRAINTS WHERE NAME = '%s' ORDER BY PKCOLSEQNUM" % (schema, table_name)
)]
cursor.close()
return primary_keys
def get_foreign_keys(self, connection, table_name, schema=None, **kw):
return []
def get_indexes(self, connection, table_name, schema=None, **kw):
return []
class SQLBase7Dialect_SA06_pyodbc(SQLBase7Dialect_SA06, PyODBCConnector):
pass
dialect = SQLBase7Dialect_SA06_pyodbc

3
sqlbase7_sa/tests/__init__.py
View file

@ -51,6 +51,5 @@ class ConnectionTestCase(TestCase):
class ReflectionTestCase(TestCase):
def runTest(self):
metadata = MetaData(bind=self.engine)
metadata.reflect(schema='SYSADM')
metadata = MetaData(bind=self.engine, reflect=True)
self.assert_(metadata.tables)