import sqlalchemy.engine as engine
import sqlalchemy.schema as schema
import sqlalchemy.ansisql as ansisql
+import sqlalchemy.types as sqltypes
from sqlalchemy.ansisql import *
from pysqlite2 import dbapi2 as sqlite
colspecs = {
- schema.INT : "INTEGER",
- schema.CHAR : "CHAR(%(length)s)",
- schema.VARCHAR : "VARCHAR(%(length)s)",
- schema.TEXT : "TEXT",
- schema.Numeric : "NUMERIC(%(precision)s, %(length)s)",
- schema.FLOAT : "NUMERIC(%(precision)s, %(length)s)",
- schema.DECIMAL : "NUMERIC(%(precision)s, %(length)s)",
- schema.TIMESTAMP : "TIMESTAMP",
- schema.DATETIME : "TIMESTAMP",
- schema.CLOB : "TEXT",
- schema.BLOB : "BLOB",
- schema.BOOLEAN : "BOOLEAN",
+ sqltypes.INT : "INTEGER",
+ sqltypes.CHAR : "CHAR(%(length)s)",
+ sqltypes.VARCHAR : "VARCHAR(%(length)s)",
+ sqltypes.TEXT : "TEXT",
+ sqltypes.Numeric : "NUMERIC(%(precision)s, %(length)s)",
+ sqltypes.FLOAT : "NUMERIC(%(precision)s, %(length)s)",
+ sqltypes.DECIMAL : "NUMERIC(%(precision)s, %(length)s)",
+ sqltypes.TIMESTAMP : "TIMESTAMP",
+ sqltypes.DATETIME : "TIMESTAMP",
+ sqltypes.CLOB : "TEXT",
+ sqltypes.BLOB : "BLOB",
+ sqltypes.BOOLEAN : "BOOLEAN",
}
pragma_names = {
- 'INTEGER' : schema.INT,
- 'VARCHAR' : schema.VARCHAR,
- 'CHAR' : schema.CHAR,
- 'TEXT' : schema.TEXT,
- 'NUMERIC' : schema.FLOAT,
- 'TIMESTAMP' : schema.TIMESTAMP,
- 'BLOB' : schema.BLOB,
+ 'INTEGER' : sqltypes.INT,
+ 'VARCHAR' : sqltypes.VARCHAR,
+ 'CHAR' : sqltypes.CHAR,
+ 'TEXT' : sqltypes.TEXT,
+ 'NUMERIC' : sqltypes.FLOAT,
+ 'TIMESTAMP' : sqltypes.TIMESTAMP,
+ 'BLOB' : sqltypes.BLOB,
}
def engine(filename, opts, **params):
class SQLiteColumnImpl(sql.ColumnSelectable):
def get_specification(self):
coltype = self.column.type
- if isinstance(coltype, types.ClassType):
+ if isinstance(coltype, type):
key = coltype
else:
key = coltype.__class__
+
colspec = self.name + " " + colspecs[key] % {'precision': getattr(coltype, 'precision', 10), 'length' : getattr(coltype, 'length', 10)}
if not self.column.nullable:
def instances(self, cursor, db = None):
result = util.HistoryArraySet()
- cursor = engine.ResultProxy(cursor, echo = db and db.echo)
+ cursor = engine.ResultProxy(cursor, echo = db and db.echo, engine = db)
imap = {}
while True:
row = cursor.fetchone()
return "ColumnProperty(%s)" % repr([hash_key(c) for c in self.columns])
def init(self, key, parent):
- print "hi im a colprop on key " + key
self.key = key
# establish a SmartProperty property manager on the object for this key
if not hasattr(parent.class_, key):
- print "Setting the class attribute"
objectstore.uow().register_attribute(parent.class_, key, uselist = False)
def execute(self, instance, row, identitykey, imap, isnew):