本文整理汇总了Python中sqlalchemy.util.callable函数的典型用法代码示例。如果您正苦于以下问题:Python callable函数的具体用法?Python callable怎么用?Python callable使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了callable函数的12个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: construct_params
def construct_params(self, params=None):
"""return a dictionary of bind parameter keys and values"""
if params:
params = util.column_dict(params)
pd = {}
for bindparam, name in self.bind_names.iteritems():
for paramname in (bindparam.key, bindparam.shortname, name):
if paramname in params:
pd[name] = params[paramname]
break
else:
if util.callable(bindparam.value):
pd[name] = bindparam.value()
else:
pd[name] = bindparam.value
return pd
else:
pd = {}
for bindparam in self.bind_names:
if util.callable(bindparam.value):
pd[self.bind_names[bindparam]] = bindparam.value()
else:
pd[self.bind_names[bindparam]] = bindparam.value
return pd
开发者ID:xihadajiang,项目名称:test,代码行数:25,代码来源:compiler.py
示例2: as_predicate
def as_predicate(cls, predicate, description=None):
if isinstance(predicate, compound):
return cls.as_predicate(predicate.fails.union(predicate.skips))
elif isinstance(predicate, Predicate):
if description and predicate.description is None:
predicate.description = description
return predicate
elif isinstance(predicate, (list, set)):
return OrPredicate(
[cls.as_predicate(pred) for pred in predicate],
description)
elif isinstance(predicate, tuple):
return SpecPredicate(*predicate)
elif isinstance(predicate, compat.string_types):
tokens = predicate.split(" ", 2)
op = spec = None
db = tokens.pop(0)
if tokens:
op = tokens.pop(0)
if tokens:
spec = tuple(int(d) for d in tokens.pop(0).split("."))
return SpecPredicate(db, op, spec, description=description)
elif util.callable(predicate):
return LambdaPredicate(predicate, description)
else:
assert False, "unknown predicate type: %s" % predicate
开发者ID:Althea-Lobo,项目名称:Flask-Skeleton,代码行数:27,代码来源:exclusions.py
示例3: process_cursor_execute
def process_cursor_execute(self, statement, parameters, context, executemany):
if not context:
return
_received_parameters = context.compiled_parameters
# recompile from the context, using the default dialect
compiled = context.compiled.statement.\
compile(dialect=DefaultDialect(), column_keys=context.compiled.column_keys)
_received_statement = re.sub(r'\n', '', str(compiled))
equivalent = self.statement == _received_statement
if self.params:
if util.callable(self.params):
params = self.params(context)
else:
params = self.params
if not isinstance(params, list):
params = [params]
# do a positive compare only
for param, received in zip(params, _received_parameters):
for k, v in param.iteritems():
if k not in received or received[k] != v:
equivalent = False
break
else:
params = {}
self._result = equivalent
if not self._result:
self._errmsg = "Testing for compiled statement %r partial params %r, " \
"received %r with params %r" % (self.statement, params, _received_statement, _received_parameters)
开发者ID:dreamwave,项目名称:rad,代码行数:35,代码来源:assertsql.py
示例4: process_cursor_execute
def process_cursor_execute(self, statement, parameters, context, executemany):
if not context:
return
_received_statement = _process_engine_statement(context.unicode_statement, context)
_received_parameters = context.compiled_parameters
# TODO: remove this step once all unit tests
# are migrated, as ExactSQL should really be *exact* SQL
sql = _process_assertion_statement(self.sql, context)
equivalent = _received_statement == sql
if self.params:
if util.callable(self.params):
params = self.params(context)
else:
params = self.params
if not isinstance(params, list):
params = [params]
equivalent = equivalent and params == context.compiled_parameters
else:
params = {}
self._result = equivalent
if not self._result:
self._errmsg = "Testing for exact statement %r exact params %r, " \
"received %r with params %r" % (sql, params, _received_statement, _received_parameters)
开发者ID:AntonNguyen,项目名称:easy_api,代码行数:29,代码来源:assertsql.py
示例5: visit_function
def visit_function(self, func, result_map=None, **kwargs):
if result_map is not None:
result_map[func.name.lower()] = (func.name, None, func.type)
name = self.function_string(func)
if util.callable(name):
return name(*[self.process(x) for x in func.clauses])
else:
return ".".join(func.packagenames + [name]) % {'expr':self.function_argspec(func)}
开发者ID:xihadajiang,项目名称:test,代码行数:10,代码来源:compiler.py
示例6: as_predicate
def as_predicate(cls, predicate):
if isinstance(predicate, Predicate):
return predicate
elif isinstance(predicate, list):
return OrPredicate([cls.as_predicate(pred) for pred in predicate])
elif isinstance(predicate, tuple):
return SpecPredicate(*predicate)
elif isinstance(predicate, basestring):
return SpecPredicate(predicate, None, None)
elif util.callable(predicate):
return LambdaPredicate(predicate)
else:
assert False, "unknown predicate type: %s" % predicate
开发者ID:NoNo1234,项目名称:the_walking_project,代码行数:13,代码来源:exclusions.py
示例7: _apply_all
def _apply_all(self, methods):
for rec in self.proxy_refs:
if rec is not None and rec.is_valid:
try:
for name in methods:
if callable(name):
name(rec)
else:
getattr(rec, name)()
except (SystemExit, KeyboardInterrupt):
raise
except Exception, e:
# fixme
sys.stderr.write("\n" + str(e) + "\n")
开发者ID:MaxMorais,项目名称:skink,代码行数:14,代码来源:engines.py
示例8: _apply_all
def _apply_all(self, methods):
# must copy keys atomically
for rec in self.proxy_refs.keys():
if rec is not None and rec.is_valid:
try:
for name in methods:
if callable(name):
name(rec)
else:
getattr(rec, name)()
except (SystemExit, KeyboardInterrupt):
raise
except Exception, e:
warnings.warn("testing_reaper couldn't close connection: %s" % e)
开发者ID:blitzmann,项目名称:Pyfa-skel,代码行数:14,代码来源:engines.py
示例9: comparator
def comparator(self):
if util.callable(self._comparator):
self._comparator = self._comparator()
if self.adapter:
self._comparator = self._comparator.adapted(self.adapter)
return self._comparator
开发者ID:MorganBorman,项目名称:cxsbs,代码行数:6,代码来源:attributes.py
示例10: visit_binary
def visit_binary(self, binary, **kwargs):
op = self.operator_string(binary.operator)
if util.callable(op):
return op(self.process(binary.left), self.process(binary.right), **binary.modifiers)
else:
return self.process(binary.left) + " " + op + " " + self.process(binary.right)
开发者ID:xihadajiang,项目名称:test,代码行数:6,代码来源:compiler.py
示例11: _instrument_class
def _instrument_class(cls):
"""Modify methods in a class and install instrumentation."""
# TODO: more formally document this as a decoratorless/Python 2.3
# option for specifying instrumentation. (likely doc'd here in code only,
# not in online docs.) Useful for C types too.
#
# __instrumentation__ = {
# 'rolename': 'methodname', # ...
# 'methods': {
# 'methodname': ('fire_{append,remove}_event', argspec,
# 'fire_{append,remove}_event'),
# 'append': ('fire_append_event', 1, None),
# '__setitem__': ('fire_append_event', 1, 'fire_remove_event'),
# 'pop': (None, None, 'fire_remove_event'),
# }
# }
# In the normal call flow, a request for any of the 3 basic collection
# types is transformed into one of our trivial subclasses
# (e.g. InstrumentedList). Catch anything else that sneaks in here...
if cls.__module__ == '__builtin__':
raise sa_exc.ArgumentError(
"Can not instrument a built-in type. Use a "
"subclass, even a trivial one.")
collection_type = util.duck_type_collection(cls)
if collection_type in __interfaces:
roles = __interfaces[collection_type].copy()
decorators = roles.pop('_decorators', {})
else:
roles, decorators = {}, {}
if hasattr(cls, '__instrumentation__'):
roles.update(copy.deepcopy(getattr(cls, '__instrumentation__')))
methods = roles.pop('methods', {})
for name in dir(cls):
method = getattr(cls, name, None)
if not util.callable(method):
continue
# note role declarations
if hasattr(method, '_sa_instrument_role'):
role = method._sa_instrument_role
assert role in ('appender', 'remover', 'iterator',
'on_link', 'converter')
roles[role] = name
# transfer instrumentation requests from decorated function
# to the combined queue
before, after = None, None
if hasattr(method, '_sa_instrument_before'):
op, argument = method._sa_instrument_before
assert op in ('fire_append_event', 'fire_remove_event')
before = op, argument
if hasattr(method, '_sa_instrument_after'):
op = method._sa_instrument_after
assert op in ('fire_append_event', 'fire_remove_event')
after = op
if before:
methods[name] = before[0], before[1], after
elif after:
methods[name] = None, None, after
# apply ABC auto-decoration to methods that need it
for method, decorator in decorators.items():
fn = getattr(cls, method, None)
if (fn and method not in methods and
not hasattr(fn, '_sa_instrumented')):
setattr(cls, method, decorator(fn))
# ensure all roles are present, and apply implicit instrumentation if
# needed
if 'appender' not in roles or not hasattr(cls, roles['appender']):
raise sa_exc.ArgumentError(
"Type %s must elect an appender method to be "
"a collection class" % cls.__name__)
elif (roles['appender'] not in methods and
not hasattr(getattr(cls, roles['appender']), '_sa_instrumented')):
methods[roles['appender']] = ('fire_append_event', 1, None)
if 'remover' not in roles or not hasattr(cls, roles['remover']):
raise sa_exc.ArgumentError(
"Type %s must elect a remover method to be "
"a collection class" % cls.__name__)
elif (roles['remover'] not in methods and
not hasattr(getattr(cls, roles['remover']), '_sa_instrumented')):
methods[roles['remover']] = ('fire_remove_event', 1, None)
if 'iterator' not in roles or not hasattr(cls, roles['iterator']):
raise sa_exc.ArgumentError(
"Type %s must elect an iterator method to be "
"a collection class" % cls.__name__)
# apply ad-hoc instrumentation from decorators, class-level defaults
# and implicit role declarations
for method, (before, argument, after) in methods.items():
setattr(cls, method,
#.........这里部分代码省略.........
开发者ID:AntonNguyen,项目名称:easy_api,代码行数:101,代码来源:collections.py
示例12: _determine_targets
def _determine_targets(self):
if isinstance(self.argument, type):
self.mapper = mapper.class_mapper(self.argument, compile=False)
elif isinstance(self.argument, mapper.Mapper):
self.mapper = self.argument
elif util.callable(self.argument):
# accept a callable to suit various deferred-configurational schemes
self.mapper = mapper.class_mapper(self.argument(), compile=False)
else:
raise sa_exc.ArgumentError(
"relation '%s' expects a class or a mapper argument (received: %s)" % (self.key, type(self.argument))
)
assert isinstance(self.mapper, mapper.Mapper), self.mapper
# accept callables for other attributes which may require deferred initialization
for attr in ("order_by", "primaryjoin", "secondaryjoin", "secondary", "_foreign_keys", "remote_side"):
if util.callable(getattr(self, attr)):
setattr(self, attr, getattr(self, attr)())
# in the case that InstrumentedAttributes were used to construct
# primaryjoin or secondaryjoin, remove the "_orm_adapt" annotation so these
# interact with Query in the same way as the original Table-bound Column objects
for attr in ("primaryjoin", "secondaryjoin"):
val = getattr(self, attr)
if val is not None:
util.assert_arg_type(val, sql.ClauseElement, attr)
setattr(self, attr, _orm_deannotate(val))
if self.order_by:
self.order_by = [expression._literal_as_column(x) for x in util.to_list(self.order_by)]
self._foreign_keys = util.column_set(
expression._literal_as_column(x) for x in util.to_column_set(self._foreign_keys)
)
self.remote_side = util.column_set(
expression._literal_as_column(x) for x in util.to_column_set(self.remote_side)
)
if not self.parent.concrete:
for inheriting in self.parent.iterate_to_root():
if inheriting is not self.parent and inheriting._get_property(self.key, raiseerr=False):
util.warn(
(
"Warning: relation '%s' on mapper '%s' supercedes "
"the same relation on inherited mapper '%s'; this "
"can cause dependency issues during flush"
)
% (self.key, self.parent, inheriting)
)
# TODO: remove 'self.table'
self.target = self.table = self.mapper.mapped_table
if self.cascade.delete_orphan:
if self.parent.class_ is self.mapper.class_:
raise sa_exc.ArgumentError(
"In relationship '%s', can't establish 'delete-orphan' cascade "
"rule on a self-referential relationship. "
"You probably want cascade='all', which includes delete cascading but not orphan detection."
% (str(self))
)
self.mapper.primary_mapper().delete_orphans.append((self.key, self.parent.class_))
开发者ID:AntonNguyen,项目名称:easy_api,代码行数:62,代码来源:properties.py
注:本文中的sqlalchemy.util.callable函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论