=======
CHANGES
=======
+0.5.0rc3
+========
+- orm
+ - "not equals" comparisons of simple many-to-one relation
+ to an instance will not drop into an EXISTS clause
+ and will compare foreign key columns instead.
+
+ - removed not-really-working use cases of comparing
+ a collection to an iterable. Use contains() to test
+ for collection membership.
+
+- sql
+ - Further simplified SELECT compilation and its relationship
+ to result row processing.
+
+ - Direct execution of a union() construct will properly set up
+ result-row processing. [ticket:1194]
+
+ - The internal notion of an "OID" or "ROWID" column has been
+ removed. It's basically not used by any dialect, and the
+ possibility of its usage with psycopg2's cursor.lastrowid
+ is basically gone now that INSERT..RETURNING is available.
+
+ - Removed "default_order_by()" method on all FromClause
+ objects.
0.5.0rc2
========
"clean" objects to better protect against operating on
objects as they're asynchronously gc'ed. [ticket:1182]
- - "not equals" comparisons of simple many-to-one relation
- to an instance will not drop into an EXISTS clause
- and will compare foreign key columns instead.
-
- - removed not-really-working use cases of comparing
- a collection to an iterable. Use contains() to test
- for collection membership.
-
- sql
- column.in_(someselect) can now be used as a columns-clause
expression without the subquery bleeding into the FROM clause
[ticket:1074]
- - Further simplified SELECT compilation and its relationship
- to result row processing.
-
- - Direct execution of a union() construct will properly set up
- result-row processing. [ticket:1194]
-
- sqlite
- Overhauled SQLite date/time bind/result processing to use
regular expressions and format strings, rather than
- Oracle will detect string-based statements which contain
comments at the front before a SELECT as SELECT statements.
[ticket:1187]
+
0.5.0rc1
========
def create_execution_context(self , *args, **kwargs):
return InfoExecutionContext(self, *args, **kwargs)
- def oid_column_name(self, column):
- return "rowid"
-
def table_names(self, connection, schema):
s = "select tabname from systables"
return [row[0] for row in connection.execute(s)]
# TODO: dont modify the original select, generate a new one
a = [ __label(c) for c in select._raw_columns ]
for c in select._order_by_clause.clauses:
- if ( __label(c) not in a ) and getattr( c , 'name' , '' ) != 'oid':
+ if ( __label(c) not in a ):
select.append_column( c )
return compiler.DefaultCompiler.visit_select(self, select)
return compiler.DefaultCompiler.visit_function( self , func )
def visit_clauselist(self, list, **kwargs):
- try:
- li = [ c for c in list.clauses if c.name != 'oid' ]
- except:
- li = [ c for c in list.clauses ]
- return ', '.join([s for s in [self.process(c) for c in li] if s is not None])
+ return ', '.join([s for s in [self.process(c) for c in list.clauses] if s is not None])
class InfoSchemaGenerator(compiler.SchemaGenerator):
def get_column_specification(self, column, first_pk=False):
def type_descriptor(self, typeobj):
return sqltypes.adapt_type(typeobj, colspecs)
- def oid_column_name(self, column):
- if not isinstance(column.table, (sql.TableClause, sql.Select)):
- return None
- else:
- return "rowid"
-
def create_xid(self):
"""create a two-phase transaction ID.
else:
return base.ResultProxy(self)
- def post_exec(self):
- if self.compiled.isinsert and self.last_inserted_ids is None:
- if not self.dialect.use_oids:
- pass
- # will raise invalid error when they go to get them
- else:
- table = self.compiled.statement.table
- if self.cursor.lastrowid is not None and table is not None and len(table.primary_key):
- s = sql.select(table.primary_key, table.oid_column == self.cursor.lastrowid)
- row = self.connection.execute(s).fetchone()
- self._last_inserted_ids = [v for v in row]
- super(PGExecutionContext, self).post_exec()
-
class PGDialect(default.DefaultDialect):
name = 'postgres'
supports_alter = True
supports_pk_autoincrement = False
default_paramstyle = 'pyformat'
- def __init__(self, use_oids=False, server_side_cursors=False, **kwargs):
+ def __init__(self, server_side_cursors=False, **kwargs):
default.DefaultDialect.__init__(self, **kwargs)
- self.use_oids = use_oids
self.server_side_cursors = server_side_cursors
def dbapi(cls):
else:
return self.context.last_inserted_ids
- def oid_column_name(self, column):
- if self.use_oids:
- return "oid"
- else:
- return None
-
def has_table(self, connection, table_name, schema=None):
# seems like case gets folded in pg_class...
if schema is None:
def create_execution_context(self, connection, **kwargs):
return SQLiteExecutionContext(self, connection, **kwargs)
- def oid_column_name(self, column):
- return "oid"
-
def is_disconnect(self, e):
return isinstance(e, self.dbapi.ProgrammingError) and "Cannot operate on a closed database." in str(e)
raise NotImplementedError()
- def oid_column_name(self, column):
- """Return the oid column name for this Dialect
-
- May return ``None`` if the dialect can't o won't support
- OID/ROWID features.
-
- The [sqlalchemy.schema#Column] instance which represents OID
- for the query being compiled is passed, so that the dialect
- can inspect the column and its parent selectable to determine
- if OID/ROWID is not selected for a particular selectable
- (i.e. Oracle doesnt support ROWID for UNION, GROUP BY,
- DISTINCT, etc.)
- """
-
- raise NotImplementedError()
-
-
def server_version_info(self, connection):
"""Return a tuple of the database's version number."""
if len(ident) > self.max_identifier_length:
raise exc.IdentifierError("Identifier '%s' exceeds maximum length of %d characters" % (ident, self.max_identifier_length))
- def oid_column_name(self, column):
- return None
-
def do_begin(self, connection):
"""Implementations might want to put logic here for turning
autocommit on/off, etc.
Strings and text() will be converted into a ``DefaultClause``
object upon initialization.
- _is_oid
- Defaults to False: used internally to indicate that this column is
- used as the quasi-hidden "oid" column
-
index
Defaults to False: indicates that this column is indexed. The name
of the index is autogenerated. to specify indexes with explicit
self.key = kwargs.pop('key', name)
self.primary_key = kwargs.pop('primary_key', False)
self.nullable = kwargs.pop('nullable', not self.primary_key)
- self._is_oid = kwargs.pop('_is_oid', False)
self.default = kwargs.pop('default', None)
self.server_default = kwargs.pop('server_default', None)
self.server_onupdate = kwargs.pop('server_onupdate', None)
self.metadata = table.metadata
if getattr(self, 'table', None) is not None:
raise exc.ArgumentError("this Column already has a table!")
- if not self._is_oid:
- self._pre_existing_column = table._columns.get(self.key)
- table._columns.replace(self)
- else:
- self._pre_existing_column = None
+ self._pre_existing_column = table._columns.get(self.key)
+ table._columns.replace(self)
if self.primary_key:
table.primary_key.replace(self)
This is used in ``Table.tometadata``.
"""
- return Column(self.name, self.type, self.default, key = self.key, primary_key = self.primary_key, nullable = self.nullable, _is_oid = self._is_oid, quote=self.quote, index=self.index, autoincrement=self.autoincrement, *[c.copy() for c in self.constraints])
+ return Column(self.name, self.type, self.default, key = self.key, primary_key = self.primary_key, nullable = self.nullable, quote=self.quote, index=self.index, autoincrement=self.autoincrement, *[c.copy() for c in self.constraints])
def _make_proxy(self, selectable, name=None):
"""Create a *proxy* for this column.
"""
fk = [ForeignKey(f._colspec) for f in self.foreign_keys]
- c = Column(name or self.name, self.type, self.default, key = name or self.key, primary_key = self.primary_key, nullable = self.nullable, _is_oid = self._is_oid, quote=self.quote, *fk)
+ c = Column(name or self.name, self.type, self.default, key = name or self.key, primary_key = self.primary_key, nullable = self.nullable, quote=self.quote, *fk)
c.table = selectable
c.proxies = [self]
c._pre_existing_column = self._pre_existing_column
- if not c._is_oid:
- selectable.columns.add(c)
- if self.primary_key:
- selectable.primary_key.add(c)
+ selectable.columns.add(c)
+ if self.primary_key:
+ selectable.primary_key.add(c)
[c._init_items(f) for f in fk]
return c
# actually present in the generated SQL
self.bind_names = {}
- # a stack. what recursive compiler doesn't have a stack ? :)
+ # stack which keeps track of nested SELECT statements
self.stack = []
# relates label names in the final SQL to
def visit_column(self, column, result_map=None, **kwargs):
- if column._is_oid:
- name = self.dialect.oid_column_name(column)
- if name is None:
- if len(column.table.primary_key) != 0:
- pk = list(column.table.primary_key)[0]
- return self.visit_column(pk, result_map=result_map, **kwargs)
- else:
- return None
- elif not column.is_literal:
+ if not column.is_literal:
name = self._truncated_identifier("colident", column.name)
else:
name = column.name
def _get_from_objects(self, **modifiers):
return []
- def default_order_by(self):
- return [self.oid_column]
-
def count(self, whereclause=None, **params):
"""return a SELECT COUNT generated against this ``FromClause``."""
col, intersect = None, None
target_set = column.proxy_set
cols = self.c
- if self.oid_column:
- cols += [self.oid_column]
for c in cols:
i = c.proxy_set.intersection(target_set)
if i and \
# from the item. this is because FromClause subclasses, when
# cloned, need to reestablish new "proxied" columns that are
# linked to the new item
- for attr in ('_columns', '_primary_key' '_foreign_keys', '_oid_column', '_embedded_columns', '_all_froms'):
+ for attr in ('_columns', '_primary_key' '_foreign_keys', '_embedded_columns', '_all_froms'):
if hasattr(self, attr):
delattr(self, attr)
columns = c = _expr_attr_func('_columns')
primary_key = _expr_attr_func('_primary_key')
foreign_keys = _expr_attr_func('_foreign_keys')
- oid_column = _expr_attr_func('_oid_column')
def _export_columns(self):
"""Initialize column collections."""
self._columns = ColumnCollection()
self._primary_key = ColumnSet()
self._foreign_keys = set()
- self._oid_column = None
self._populate_column_collection()
def _populate_column_collection(self):
supports_execution = True
_hide_froms = []
- oid_column = None
def __init__(self, text = "", bind=None, bindparams=None, typemap=None, autocommit=False):
self._bind = bind
(c for c in columns if c.primary_key), self.onclause))
self._columns.update((col._label, col) for col in columns)
self._foreign_keys.update(itertools.chain(*[col.foreign_keys for col in columns]))
- self._oid_column = self.left.oid_column
def _copy_internals(self, clone=_clone):
self._reset_exported()
def _populate_column_collection(self):
for col in self.element.columns:
col._make_proxy(self)
- if self.element.oid_column is not None:
- self._oid_column = self.element.oid_column._make_proxy(self)
def _copy_internals(self, clone=_clone):
self._reset_exported()
``_ColumnClause``.
"""
- def __init__(self, text, selectable=None, type_=None, _is_oid=False, is_literal=False):
+ def __init__(self, text, selectable=None, type_=None, is_literal=False):
ColumnElement.__init__(self)
self.key = self.name = text
self.table = selectable
self.type = sqltypes.to_instance(type_)
- self._is_oid = _is_oid
self.__label = None
self.is_literal = is_literal
# propigate the "is_literal" flag only if we are keeping our name,
# otherwise its considered to be a label
is_literal = self.is_literal and (name is None or name == self.name)
- c = _ColumnClause(name or self.name, selectable=selectable, _is_oid=self._is_oid, type_=self.type, is_literal=is_literal)
+ c = _ColumnClause(name or self.name, selectable=selectable, type_=self.type, is_literal=is_literal)
c.proxies = [self]
- if attach and not self._is_oid:
+ if attach:
selectable.columns[c.name] = c
return c
def __init__(self, name, *columns):
super(TableClause, self).__init__()
self.name = self.fullname = name
- self._oid_column = _ColumnClause('oid', self, _is_oid=True)
self._columns = ColumnCollection()
self._primary_key = ColumnSet()
self._foreign_keys = set()
proxy = cols[0]._make_proxy(self, name=self.use_labels and cols[0]._label or None)
proxy.proxies = cols
- oid_proxies = [
- c for c in [f.oid_column for f in self.selects] if c is not None
- ]
-
- if oid_proxies:
- col = oid_proxies[0]._make_proxy(self)
- col.proxies = oid_proxies
- self._oid_column = col
-
def _copy_internals(self, clone=_clone):
self._reset_exported()
self.selects = [clone(s) for s in self.selects]
def _populate_column_collection(self):
for c in self.__exportable_columns():
c._make_proxy(self, name=self.use_labels and c._label or None)
-
- oid_proxies = [c for c in
- [f.oid_column for f in self.locate_all_froms()
- if f is not self] if c is not None
- ]
-
- if oid_proxies:
- col = oid_proxies[0]._make_proxy(self)
- col.proxies = oid_proxies
- self._oid_column = col
def self_group(self, against=None):
"""return a 'grouping' construct as per the ClauseElement specification.
query = table1.join(table2, table1.c.myid==table2.c.otherid).outerjoin(table3, table3.c.userid==table2.c.otherid)
- self.assert_compile(query.select().order_by(table1.oid_column).limit(10).offset(5),
+ self.assert_compile(query.select().order_by(table1.c.name).limit(10).offset(5),
"SELECT myid, name, description, otherid, othername, userid, "
"otherstuff FROM (SELECT /*+ FIRST_ROWS(10) */ myid, name, description, "
"myothertable.otherid AS otherid, myothertable.othername AS othername, "
"thirdtable.userid AS userid, thirdtable.otherstuff AS otherstuff FROM mytable, "
"myothertable, thirdtable WHERE thirdtable.userid(+) = myothertable.otherid AND "
- "mytable.myid = myothertable.otherid ORDER BY mytable.rowid) WHERE "
+ "mytable.myid = myothertable.otherid ORDER BY mytable.name) WHERE "
"ROWNUM <= :ROWNUM_1) WHERE ora_rn > :ora_rn_1", dialect=oracle.dialect(use_ansi=False))
def test_alias_outer_join(self):
s = select([at_alias, addresses]).\
select_from(addresses.outerjoin(at_alias, addresses.c.address_type_id==at_alias.c.id)).\
where(addresses.c.user_id==7).\
- order_by(addresses.oid_column, address_types.oid_column)
+ order_by(addresses.c.id, address_types.c.id)
self.assert_compile(s, "SELECT address_types_1.id, address_types_1.name, addresses.id, addresses.user_id, "
"addresses.address_type_id, addresses.email_address FROM addresses LEFT OUTER JOIN address_types address_types_1 "
- "ON addresses.address_type_id = address_types_1.id WHERE addresses.user_id = :user_id_1 ORDER BY addresses.rowid, "
- "address_types.rowid")
+ "ON addresses.address_type_id = address_types_1.id WHERE addresses.user_id = :user_id_1 ORDER BY addresses.id, "
+ "address_types.id")
class MultiSchemaTest(TestBase, AssertsCompiledSQL):
"""instructions:
mapper(Address, addresses, properties={
'phones': relation(Phone, lazy=False, backref='address',
- order_by=phone_numbers.default_order_by())})
+ order_by=phone_numbers.c.phone_id)})
mapper(Company, companies, properties={
'addresses': relation(Address, lazy=False, backref='company',
- order_by=addresses.default_order_by())})
+ order_by=addresses.c.address_id)})
mapper(Item, items)
mapper(Invoice, invoices, properties={
'items': relation(Item, lazy=False, backref='invoice',
- order_by=items.default_order_by()),
+ order_by=items.c.item_id),
'company': relation(Company, lazy=False, backref='invoices')})
c1 = Company(company_name='company 1', addresses=[
users.c.id == addresses.c.user_id,
group_by=[c for c in users.c]).alias('myselect')
- mapper(User, s, order_by=s.default_order_by())
+ mapper(User, s, order_by=s.c.id)
sess = create_session()
l = sess.query(User).all()
mapper(User, users, properties=dict(
addresses=relation(Address, lazy=False,
- order_by=addresses.default_order_by()),
+ order_by=addresses.c.id),
orders=relation(Order, lazy=False,
- order_by=orders.default_order_by())))
+ order_by=orders.c.id)))
sess = create_session()
mapper(Item, items, properties=dict(
keywords=relation(Keyword, item_keywords,
- order_by=item_keywords.default_order_by())))
+ order_by=item_keywords.c.item_id)))
mapper(Order, orders, properties=dict(
items=relation(Item, order_items,
- order_by=items.default_order_by())))
+ order_by=items.c.id)))
- mapper(User, users, order_by=users.default_order_by(), properties=dict(
- orders=relation(Order, order_by=orders.default_order_by())))
+ mapper(User, users, order_by=users.c.id, properties=dict(
+ orders=relation(Order, order_by=orders.c.id)))
@testing.resolve_artifact_names
def test_deep_options_1(self):
def test_profile_1a_populate(self):
self.test_baseline_1a_populate()
- @profiling.function_call_count(1254, {'2.4':1184})
+ @profiling.function_call_count(1185, {'2.4':1184})
def test_profile_2_insert(self):
self.test_baseline_2_insert()
print x.execute().fetchall()
- def test_oid(self):
- """test that a primary key column compiled as the 'oid' column gets proper length truncation"""
- from sqlalchemy.databases import postgres
- dialect = postgres.PGDialect()
- dialect.max_identifier_length = 30
- tt = table1.select(use_labels=True).alias('foo')
- x = select([tt], use_labels=True, order_by=tt.oid_column).compile(dialect=dialect)
- #print x
- # assert it doesnt end with "ORDER BY foo.some_large_named_table_this_is_the_primarykey_column"
- assert str(x).endswith("""ORDER BY foo.some_large_named_table_t_2""")
-
if __name__ == '__main__':
testenv.main()
assert u1.corresponding_column(table2.c.otherid) is u1.c.myid
- assert u1.corresponding_column(table1.oid_column) is u1.oid_column
- assert u1.corresponding_column(table2.oid_column) is u1.oid_column
-
# TODO - why is there an extra space before the LIMIT ?
self.assert_compile(
union(
UNION SELECT mytable.myid FROM mytable"
)
- # test unions working with non-oid selectables
s = select([column('foo'), column('bar')])
s = union(s, s)
s = union(s, s)
def test_singular_union(self):
u = union(select([table.c.col1, table.c.col2, table.c.col3]), select([table.c.col1, table.c.col2, table.c.col3]))
- assert u.oid_column is not None
u = union(select([table.c.col1, table.c.col2, table.c.col3]))
- assert u.oid_column
assert u.c.col1
assert u.c.col2
assert u.c.col3
assert j4.corresponding_column(j2.c.aid) is j4.c.aid
assert j4.corresponding_column(a.c.id) is j4.c.id
- @testing.emits_warning('.*replaced by another column with the same key')
- def test_oid(self):
- # the oid column of a selectable currently proxies all
- # oid columns found within.
- s = table.select()
- s2 = table2.select()
- s3 = select([s, s2])
- assert s3.corresponding_column(table.oid_column) is s3.oid_column
- assert s3.corresponding_column(table2.oid_column) is s3.oid_column
- assert s3.corresponding_column(s.oid_column) is s3.oid_column
- assert s3.corresponding_column(s2.oid_column) is s3.oid_column
-
- u = s.union(s2)
- assert u.corresponding_column(table.oid_column) is u.oid_column
- assert u.corresponding_column(table2.oid_column) is u.oid_column
- assert u.corresponding_column(s.oid_column) is u.oid_column
- assert u.corresponding_column(s2.oid_column) is u.oid_column
-
def test_two_metadata_join_raises(self):
m = MetaData()
m2 = MetaData()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
