- all "type" keyword arguments, such as those to bindparam(), column(),
Column(), and func.<something>(), renamed to "type_". those objects
still name their "type" attribute as "type".
-
+
+ - case_sensitive=(True|False) setting removed from schema items, since
+ checking this state added a lot of method call overhead and there was
+ no decent reason to ever set it to False. Table and column names which are
+ all lower case will be treated as case-insenstive (yes we adjust for
+ Oracle's UPPERCASE style too).
+
- transactions:
- added context manager (with statement) support for transactions
- added support for two phase commit, works with mysql and postgres so far.
- table reflection can now be performed in as little as one round-trip
- ANSI and ANSI_QUOTES sql modes are now supported
- indexes are now reflected
+
- postgres
- Added PGArray datatype for using postgres array datatypes
def _illegal_initial_characters(self):
return ILLEGAL_INITIAL_CHARACTERS
- def _requires_quotes(self, value, case_sensitive):
+ def _requires_quotes(self, value):
"""Return True if the given identifier requires quoting."""
return \
value in self._reserved_words() \
or (value[0] in self._illegal_initial_characters()) \
or bool(len([x for x in unicode(value) if x not in self._legal_characters()])) \
- or (case_sensitive and value.lower() != value)
+ or (value.lower() != value)
def __generic_obj_format(self, obj, ident):
if getattr(obj, 'quote', False):
return self.quote_identifier(ident)
if self.dialect.cache_identifiers:
- case_sens = getattr(obj, 'case_sensitive', None)
try:
- return self.__strings[(ident, case_sens)]
+ return self.__strings[ident]
except KeyError:
- if self._requires_quotes(ident, getattr(obj, 'case_sensitive', ident == ident.lower())):
- self.__strings[(ident, case_sens)] = self.quote_identifier(ident)
+ if self._requires_quotes(ident):
+ self.__strings[ident] = self.quote_identifier(ident)
else:
- self.__strings[(ident, case_sens)] = ident
- return self.__strings[(ident, case_sens)]
+ self.__strings[ident] = ident
+ return self.__strings[ident]
else:
- if self._requires_quotes(ident, getattr(obj, 'case_sensitive', ident == ident.lower())):
+ if self._requires_quotes(ident):
return self.quote_identifier(ident)
else:
return ident
def should_quote(self, object):
- return object.quote or self._requires_quotes(object.name, object.case_sensitive)
+ return object.quote or self._requires_quotes(object.name)
def format_sequence(self, sequence):
return self.__generic_obj_format(sequence, sequence.name)
def _fold_identifier_case(self, value):
return value.lower()
- def _requires_quotes(self, value, case_sensitive):
+ def _requires_quotes(self, value):
return False
class InfoSchemaDroper(ansisql.ANSISchemaDropper):
m = self._derived_metadata()
return m and m.bind or None
- def _set_casing_strategy(self, kwargs, keyname='case_sensitive'):
- """Set the "case_sensitive" argument sent via keywords to the item's constructor.
-
- For the purposes of Table's 'schema' property, the name of the
- variable is optionally configurable.
- """
- setattr(self, '_%s_setting' % keyname, kwargs.pop(keyname, None))
-
- def _determine_case_sensitive(self, keyname='case_sensitive'):
- """Determine the `case_sensitive` value for this item.
-
- For the purposes of Table's `schema` property, the name of the
- variable is optionally configurable.
-
- A local non-None value overrides all others. After that, the
- parent item (i.e. ``Column`` for a ``Sequence``, ``Table`` for
- a ``Column``, ``MetaData`` for a ``Table``) is searched for a
- non-None setting, traversing each parent until none are found.
- finally, case_sensitive is set to True as a default.
- """
-
- local = getattr(self, '_%s_setting' % keyname, None)
- if local is not None:
- return local
- parent = self
- while parent is not None:
- parent = parent._get_parent()
- if parent is not None:
- parentval = getattr(parent, '_case_sensitive_setting', None)
- if parentval is not None:
- return parentval
- return True
-
- def _get_case_sensitive(self):
- """late-compile the 'case-sensitive' setting when first accessed.
-
- typically the SchemaItem will be assembled into its final structure
- of other SchemaItems at this point, whereby it can attain this setting
- from its containing SchemaItem if not defined locally.
- """
-
- try:
- return self.__case_sensitive
- except AttributeError:
- self.__case_sensitive = self._determine_case_sensitive()
- return self.__case_sensitive
- case_sensitive = property(_get_case_sensitive)
metadata = property(lambda s:s._derived_metadata())
bind = property(lambda s:s._get_bind())
the database. This flag overrides all other quoting
behavior.
- case_sensitive
- Defaults to True: indicates quoting should be used if the
- identifier contains mixed case.
-
- case_sensitive_schema
- Defaults to True: indicates quoting should be used if the
- identifier contains mixed case.
"""
super(Table, self).__init__(name)
self._metadata = metadata
self.fullname = self.name
self.owner = kwargs.pop('owner', None)
- self._set_casing_strategy(kwargs)
- self._set_casing_strategy(kwargs, keyname='case_sensitive_schema')
-
if len([k for k in kwargs if not re.match(r'^(?:%s)_' % '|'.join(databases.__all__), k)]):
raise TypeError("Invalid argument(s) for Table: %s" % repr(kwargs.keys()))
# implement it differently
pass
- def _get_case_sensitive_schema(self):
- try:
- return getattr(self, '_case_sensitive_schema')
- except AttributeError:
- setattr(self, '_case_sensitive_schema', self._determine_case_sensitive(keyname='case_sensitive_schema'))
- return getattr(self, '_case_sensitive_schema')
- case_sensitive_schema = property(_get_case_sensitive_schema)
-
def _set_primary_key(self, pk):
if getattr(self, '_primary_key', None) in self.constraints:
self.constraints.remove(self._primary_key)
as dialects can auto-detect conditions where quoting is
required.
- case_sensitive
- Defaults to True: indicates quoting should be used if the
- identifier contains mixed case.
"""
super(Column, self).__init__(name, None, type_)
self.index = kwargs.pop('index', None)
self.unique = kwargs.pop('unique', None)
self.quote = kwargs.pop('quote', False)
- self._set_casing_strategy(kwargs)
self.onupdate = kwargs.pop('onupdate', None)
self.autoincrement = kwargs.pop('autoincrement', True)
self.constraints = util.Set()
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, case_sensitive=self._case_sensitive_setting, quote=self.quote, index=self.index, *[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, _is_oid = self._is_oid, quote=self.quote, index=self.index, *[c.copy() for c in self.constraints])
def _make_proxy(self, selectable, name = None):
"""Create a *proxy* for this column.
return c
- def _case_sens(self):
- """Redirect the `case_sensitive` accessor to use the ultimate
- parent column which created this one."""
-
- return self.__originating_column._get_case_sensitive()
- case_sensitive = property(_case_sens, lambda s,v:None)
-
def get_children(self, schema_visitor=False, **kwargs):
if schema_visitor:
return [x for x in (self.default, self.onupdate) if x is not None] + \
self.increment = increment
self.optional=optional
self.quote = quote
- self._set_casing_strategy(kwargs)
def __repr__(self):
return "Sequence(%s)" % ', '.join(
__visit_name__ = 'metadata'
- def __init__(self, bind=None, reflect=False, case_sensitive=None):
+ def __init__(self, bind=None, reflect=False):
"""Create a new MetaData object.
bind
set. For finer control over loaded tables, use the ``reflect``
method of ``MetaData``.
- case_sensitive
- A default case sensitive setting for all contained objects.
- Defaults to sensitive.
"""
self.tables = {}
- self._set_casing_strategy({'case_sensitive': case_sensitive})
self.bind = bind
if reflect:
if not bind:
return key in self.tables
def __getstate__(self):
- return {'tables': self.tables,
- 'casesensitive': self._case_sensitive_setting}
+ return {'tables': self.tables}
def __setstate__(self, state):
self.tables = state['tables']
- self._case_sensitive_setting = state['casesensitive']
self._bind = None
def is_bound(self):
alias = '{ANON %d %s}' % (id(self), alias or 'anon')
self.name = alias
self.encodedname = alias.encode('ascii', 'backslashreplace')
- self.case_sensitive = getattr(baseselectable, "case_sensitive", True)
def is_derived_from(self, fromclause):
x = self.selectable
self.name = name or "{ANON %d %s}" % (id(self), getattr(obj, 'name', 'anon'))
self.obj = obj.self_group(against=Operators.as_)
- self.case_sensitive = getattr(obj, "case_sensitive", True)
self.type = sqltypes.to_instance(type_ or getattr(obj, 'type', None))
key = property(lambda s: s.name)
``TypeEngine`` object which can associate this ``_ColumnClause``
with a type.
- case_sensitive
- defines whether identifier quoting rules will be applied to the
- generated text of this ``_ColumnClause`` so that it is identified in
- a case-sensitive manner.
-
is_literal
if True, the ``_ColumnClause`` is assumed to be an exact expression
that will be delivered to the output with no quoting rules applied
"""
- def __init__(self, text, selectable=None, type_=None, _is_oid=False, case_sensitive=True, is_literal=False):
+ def __init__(self, text, selectable=None, type_=None, _is_oid=False, is_literal=False):
self.key = self.name = text
self.encodedname = isinstance(self.name, unicode) and self.name.encode('ascii', 'backslashreplace') or self.name
self.table = selectable
self._is_oid = _is_oid
self._distance = 0
self.__label = None
- self.case_sensitive = case_sensitive
self.is_literal = is_literal
def _clone(self):
print res2
assert(res2==[(1,2,3),(2,2,3),(4,3,2)])
- def testcascade(self):
- lcmetadata = MetaData(case_sensitive=False)
- t1 = Table('SomeTable', lcmetadata,
- Column('UcCol', Integer),
- Column('normalcol', String))
- t2 = Table('othertable', lcmetadata,
- Column('UcCol', Integer),
- Column('normalcol', String, ForeignKey('SomeTable.normalcol')))
- assert lcmetadata.case_sensitive is False
- assert t1.c.UcCol.case_sensitive is False
- assert t2.c.normalcol.case_sensitive is False
-
@testing.unsupported('oracle')
def testlabels(self):
"""test the quoting of labels.
x = x.select()
assert str(x) == '''SELECT "SomeLabel" \nFROM (SELECT 'FooCol' AS "SomeLabel" \nFROM "ImATable")'''
- # oracle doesn't support non-case-sensitive until ticket #726 is fixed
- @testing.unsupported('oracle')
- def testlabelsnocase(self):
- metadata = MetaData()
- table1 = Table('SomeCase1', metadata,
- Column('lowercase', Integer, primary_key=True),
- Column('UPPERCASE', Integer),
- Column('MixedCase', Integer))
- table2 = Table('SomeCase2', metadata,
- Column('id', Integer, primary_key=True, key='d123'),
- Column('col2', Integer, key='u123'),
- Column('MixedCase', Integer))
-
- # first test case sensitive tables migrating via tometadata
- meta = MetaData(testbase.db, case_sensitive=False)
- lc_table1 = table1.tometadata(meta)
- lc_table2 = table2.tometadata(meta)
- assert lc_table1.case_sensitive is False
- assert lc_table1.c.UPPERCASE.case_sensitive is False
- s = lc_table1.select()
- assert hasattr(s.c.UPPERCASE, "case_sensitive")
- assert s.c.UPPERCASE.case_sensitive is False
-
- # now, the aliases etc. should be case-insensitive. PG will screw up if this doesnt work.
- # also, if this test is run in the context of the other tests, we also test that the dialect properly
- # caches identifiers with "case_sensitive" and "not case_sensitive" separately.
- meta.create_all()
- try:
- x = lc_table1.select(distinct=True).alias("lala").select().scalar()
- finally:
- meta.drop_all()
class PreparerTest(PersistTest):
"""Test the db-agnostic quoting services of ANSIIdentifierPreparer."""