# If it's decided that issuing that sort of SQL leaves you SOL, then
# this can prefer the driver value.
rs = connection.execute("SHOW VARIABLES LIKE 'character_set%%'")
- opts = dict([(row[0], row[1]) for row in self._compat_fetchall(rs)])
+ opts = {row[0]: row[1] for row in self._compat_fetchall(rs)}
for key in ('character_set_connection', 'character_set'):
if opts.get(key, None):
return opts[key]
# If it's decided that issuing that sort of SQL leaves you SOL, then
# this can prefer the driver value.
rs = connection.execute("SHOW VARIABLES LIKE 'character_set%%'")
- opts = dict((row[0], row[1]) for row in self._compat_fetchall(rs))
+ opts = {row[0]: row[1] for row in self._compat_fetchall(rs)}
for key in ('character_set_connection', 'character_set'):
if opts.get(key, None):
return opts[key]
class OracleIdentifierPreparer(compiler.IdentifierPreparer):
- reserved_words = set([x.lower() for x in RESERVED_WORDS])
- illegal_initial_characters = set(
- (str(dig) for dig in range(0, 10))).union(["_", "$"])
+ reserved_words = {x.lower() for x in RESERVED_WORDS}
+ illegal_initial_characters = {str(dig) for dig in range(0, 10)} \
+ .union(["_", "$"])
def _bindparam_requires_quotes(self, value):
"""Return True if the given identifier requires quoting."""
oracle_sys_col = re.compile(r'SYS_NC\d+\$', re.IGNORECASE)
def upper_name_set(names):
- return set([i.upper() for i in names])
+ return {i.upper() for i in names}
pk_names = upper_name_set(pkeys)
if additional_tests:
tests += additional_tests
- results = set([check_unicode(test) for test in tests])
+ results = {check_unicode(test) for test in tests}
if results.issuperset([True, False]):
return "conditional"
else:
- return results == set([True])
+ return results == {True}
def _check_unicode_description(self, connection):
# all DBAPIs on Py2K return cursor.description as encoded,
self.parameters = parameters
else:
self.parameters = [
- dict((dialect._encoder(k)[0], d[k]) for k in d)
+ {dialect._encoder(k)[0]: d[k] for k in d}
for d in parameters
] or [{}]
else:
query = (
len(tokens) > 1 and dict(util.parse_qsl(tokens[1]))) or None
if util.py2k and query is not None:
- query = dict((k.encode('ascii'), query[k]) for k in query)
+ query = {k.encode('ascii'): query[k] for k in query}
else:
query = None
components['query'] = query
)
o2m_kws = {}
- nullable = False not in set([fk.parent.nullable for fk in fks])
+ nullable = False not in {fk.parent.nullable for fk in fks}
if not nullable:
o2m_kws['cascade'] = "all, delete-orphan"
.. versionadded:: 1.0.5
"""
- return set([attribute.key])
+ return {attribute.key}
@classmethod
def _listen_on_attribute(cls, attribute, coerce, parent_cls):
@classmethod
def _get_listen_keys(cls, attribute):
- return set([attribute.key]).union(attribute.property._attribute_keys)
+ return {attribute.key}.union(attribute.property._attribute_keys)
def changed(self):
"""Subclasses should call this method whenever change events occur."""
{
tablea.col1:
- set([tableb.col1, tablec.col1]),
+ {tableb.col1, tablec.col1},
tablea.col2:
- set([tabled.col2])
+ {tabled.col2}
}
"""
@_memoized_configured_property
def _primary_key_propkeys(self):
- return set([prop.key for prop in self._all_pk_props])
+ return {prop.key for prop in self._all_pk_props}
def _get_state_attr_by_column(
self, state, dict_, column,
search_keys = mapper._primary_key_propkeys
if mapper._version_id_prop:
- search_keys = set([mapper._version_id_prop.key]).union(search_keys)
+ search_keys = {mapper._version_id_prop.key}.union(search_keys)
def _changed_dict(mapper, state):
return dict(
self._gather_columns_with_annotation(
self.secondaryjoin, annotation)
)
- return set([x._deannotate() for x in s])
+ return {x._deannotate() for x in s}
def _gather_columns_with_annotation(self, clause, *annotation):
annotation = set(annotation)
secondaryjoin, {}, col_to_bind)
lazywhere = sql.and_(lazywhere, secondaryjoin)
- bind_to_col = dict((binds[col].key, col) for col in binds)
+ bind_to_col = {binds[col].key: col for col in binds}
return lazywhere, bind_to_col, equated_columns
'using', 'verbose', 'when', 'where'])
LEGAL_CHARACTERS = re.compile(r'^[A-Z0-9_$]+$', re.I)
-ILLEGAL_INITIAL_CHARACTERS = set([str(x) for x in range(0, 10)]).union(['$'])
+ILLEGAL_INITIAL_CHARACTERS = {str(x) for x in range(0, 10)}.union(['$'])
BIND_PARAMS = re.compile(r'(?<![:\w\$\x5c]):([\w\$]+)(?![:\w\$])', re.UNICODE)
BIND_PARAMS_ESC = re.compile(r'\x5c(:[\w\$]*)(?![:\w\$])', re.UNICODE)
toplevel = not self.stack
self.stack.append(
- {'correlate_froms': set([update_stmt.table]),
- "asfrom_froms": set([update_stmt.table]),
+ {'correlate_froms': {update_stmt.table},
+ "asfrom_froms": {update_stmt.table},
"selectable": update_stmt})
extra_froms = update_stmt._extra_froms
def visit_delete(self, delete_stmt, asfrom=False, **kw):
toplevel = not self.stack
- self.stack.append({'correlate_froms': set([delete_stmt.table]),
- "asfrom_froms": set([delete_stmt.table]),
+ self.stack.append({'correlate_froms': {delete_stmt.table},
+ "asfrom_froms": {delete_stmt.table},
"selectable": delete_stmt})
crud._setup_crud_params(self, delete_stmt, crud.ISDELETE, **kw)
self.parameters, self._has_multi_parameters = \
self._process_colparams(
- dict((_column_as_key(n), Null()) for n in names))
+ {_column_as_key(n): Null() for n in names})
self.select_names = names
self.inline = True
# TODO: this could be made memoized
# if the memoization is reset on each generative call.
froms = []
- seen = set([self.table])
+ seen = {self.table}
if self._whereclause is not None:
for item in _from_objects(self._whereclause):
raise NotImplementedError()
-_commutative = set([eq, ne, add, mul])
+_commutative = {eq, ne, add, mul}
-_comparison = set([eq, ne, lt, gt, ge, le, between_op, like_op])
+_comparison = {eq, ne, lt, gt, ge, le, between_op, like_op}
def is_comparison(op):
columns = cols_w_table
- tables = set([c.table for c in columns])
+ tables = {c.table for c in columns}
if len(tables) == 1:
self._set_parent_with_dispatch(tables.pop())
elif len(tables) > 1 and not self._allow_multiple_tables:
replacement by a given replacement function."""
cloned = {}
- stop_on = set([id(x) for x in opts.get('stop_on', [])])
+ stop_on = {id(x) for x in opts.get('stop_on', [])}
def clone(elem, **kw):
if id(elem) in stop_on or \
eq_(c.type.length, reflected_c.type.length)
eq_(
- set([f.column.name for f in c.foreign_keys]),
- set([f.column.name for f in reflected_c.foreign_keys])
+ {f.column.name for f in c.foreign_keys},
+ {f.column.name for f in reflected_c.foreign_keys}
)
if c.server_default:
assert isinstance(reflected_c.server_default,
return id(self)
found = util.IdentitySet(result)
- expected = set([immutabledict(e) for e in expected])
+ expected = {immutabledict(e) for e in expected}
for wrong in util.itertools_filterfalse(lambda o:
isinstance(o, cls), found):
"select u.username from all_users u where username "
"like 'TEST_%' and not exists (select username "
"from v$session where username=u.username)")
- all_names = set([username.lower() for (username, ) in to_reap])
+ all_names = {username.lower() for (username, ) in to_reap}
to_drop = set()
for name in all_names:
if name.endswith("_ts1") or name.endswith("_ts2"):
def Table(*args, **kw):
"""A schema.Table wrapper/hook for dialect-specific tweaks."""
- test_opts = dict([(k, kw.pop(k)) for k in list(kw)
- if k.startswith('test_')])
+ test_opts = {k: kw.pop(k) for k in list(kw) if k.startswith('test_')}
kw.update(table_options)
def Column(*args, **kw):
"""A schema.Column wrapper/hook for dialect-specific tweaks."""
- test_opts = dict([(k, kw.pop(k)) for k in list(kw)
- if k.startswith('test_')])
+ test_opts = {k: kw.pop(k) for k in list(kw) if k.startswith('test_')}
if not config.requirements.foreign_key_ddl.enabled_for_config(config):
args = [arg for arg in args if not isinstance(arg, schema.ForeignKey)]
('dingalings', 'dingaling_id'),
]:
cols = insp.get_columns(tname)
- id_ = dict((c['name'], c) for c in cols)[cname]
+ id_ = {c['name']: c for c in cols}[cname]
assert id_.get('autoincrement', True)
t.create()
t.insert().execute([{'x': x} for x in input_])
- result = set([row[0] for row in t.select().execute()])
+ result = {row[0] for row in t.select().execute()}
output = set(output)
if filter_:
result = set(filter_(x) for x in result)
Useful for asserting the results of an unordered query.
"""
- return set([tuple(row) for row in results])
+ return {tuple(row) for row in results}
def fail(msg):
.. versionadded:: 0.8
"""
- return dict((key, self.__dict__[key]) for key in self.keys())
+ return {key: self.__dict__[key] for key in self.keys()}
class _LW(AbstractKeyedTuple):
if isinstance(cls, types.ClassType):
return list()
- hier = set([cls])
+ hier = {cls}
process = list(cls.__mro__)
while process:
c = process.pop()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
