def deregister_response(fn):
""" Deregister response from the registry.
It's a decorator.
"""
@wraps(fn)
def inner(self, *args, **kwargs):
item_or_request = fn(self, *args, **kwargs)
if isinstance(item_or_request, types.GeneratorType):
item_or_request = get_consistent_generator(item_or_request)
# Only decrease counter if the item_or_request passed the filter
if item_or_request:
response = self._get_response(args, kwargs)
self._decrease_counter(response)
return item_or_request
return inner
python类GeneratorType()的实例源码
def visit(self, node):
stack = [ Visit(node) ]
last_result = None
while stack:
try:
last = stack[-1]
if isinstance(last, types.GeneratorType):
stack.append(last.send(last_result))
last_result = None
elif isinstance(last, Visit):
stack.append(self._visit(stack.pop().node))
else:
last_result = stack.pop()
except StopIteration:
stack.pop()
return last_result
def visit(self, node):
stack = [ node ]
last_result = None
while stack:
try:
last = stack[-1]
if isinstance(last, types.GeneratorType):
stack.append(last.send(last_result))
last_result = None
elif isinstance(last, Node):
stack.append(self._visit(stack.pop()))
else:
last_result = stack.pop()
except StopIteration:
stack.pop()
return last_result
def get_type(obj):
"""Return the static type that would be used in a type hint"""
if isinstance(obj, type):
return Type[obj]
elif isinstance(obj, _BUILTIN_CALLABLE_TYPES):
return Callable
elif isinstance(obj, types.GeneratorType):
return Iterator[Any]
typ = type(obj)
if typ is list:
elem_type = shrink_types(get_type(e) for e in obj)
return List[elem_type]
elif typ is set:
elem_type = shrink_types(get_type(e) for e in obj)
return Set[elem_type]
elif typ is dict:
key_type = shrink_types(get_type(k) for k in obj.keys())
val_type = shrink_types(get_type(v) for v in obj.values())
return Dict[key_type, val_type]
elif typ is tuple:
if not obj:
return Tuple
return Tuple[tuple(get_type(e) for e in obj)]
return typ
def add(self, *objs):
"""
Add a sequence of polynomials or containers of polynomials.
Example
-------
>>> from sympy.polys.rings import ring
>>> from sympy.polys.domains import ZZ
>>> R, x = ring("x", ZZ)
>>> R.add([ x**2 + 2*i + 3 for i in range(4) ])
4*x**2 + 24
>>> _.factor_list()
(4, [(x**2 + 6, 1)])
"""
p = self.zero
for obj in objs:
if is_sequence(obj, include=GeneratorType):
p += self.add(*obj)
else:
p += obj
return p
def mul(self, *objs):
"""
Multiply a sequence of polynomials or containers of polynomials.
Example
-------
>>> from sympy.polys.rings import ring
>>> from sympy.polys.domains import ZZ
>>> R, x = ring("x", ZZ)
>>> R.mul([ x**2 + 2*i + 3 for i in range(4) ])
x**8 + 24*x**6 + 206*x**4 + 744*x**2 + 945
>>> _.factor_list()
(1, [(x**2 + 3, 1), (x**2 + 5, 1), (x**2 + 7, 1), (x**2 + 9, 1)])
"""
p = self.one
for obj in objs:
if is_sequence(obj, include=GeneratorType):
p *= self.mul(*obj)
else:
p *= obj
return p
def isgenerator(object):
"""Return true if the object is a generator.
Generator objects provide these attributes:
__iter__ defined to support interation over container
close raises a new GeneratorExit exception inside the
generator to terminate the iteration
gi_code code object
gi_frame frame object or possibly None once the generator has
been exhausted
gi_running set to 1 when generator is executing, 0 otherwise
next return the next item from the container
send resumes the generator and "sends" a value that becomes
the result of the current yield-expression
throw used to raise an exception inside the generator"""
return isinstance(object, types.GeneratorType)
def run(self):
threadless.log.debug("threadlet: %s: tasklet(%s): running", self.threadlet.name, self.name)
self.running = True
try:
value = self.func(self)
if isinstance(value, types.GeneratorType):
value = yield from value
threadless.log.debug("threadlet: %s: tasklet(%s): done", self.threadlet.name, self.name)
except asyncio.CancelledError:
threadless.log.warn("threadlet: %s: tasklet(%s): cancelled", self.threadlet.name, self.name)
except Exception:
threadless.log.exception("threadlet: %s: tasklet(%s): exception", self.threadlet.name, self.name)
del self.running
if self.suspended or self.cancelled:
return
if self.period and self not in self.threadlet.timeouts:
self.schedule(self.period)
def work(self, queues, **kwargs):
# ??????queue??????
self.command_queue = command_queue = queues['command_queue']
self.data_queue = data_queue = queues['data_queue']
while True:
# ????
cmd = command_queue.get()
# ??????????
if cmd:
raw_datas = self.user_check(cmd)
if isinstance(raw_datas, types.GeneratorType):
for raw_data in raw_datas:
# ?????????handler
data_queue.put(raw_data)
else:
# ?????????handler
data_queue.put(raw_datas)
else:
log.error('\nNo command received')
kwargs['record'].thread_signal[kwargs['name']] = time.time()
def content(self):
content = self._fixed_content
if not content and self.parent:
content = self.parent._find_content(self._name)
if isinstance(content, DOMElement) or content:
if isinstance(content, DOMElement):
yield content
elif isinstance(content, (list, tuple, GeneratorType)):
yield from content
elif isinstance(content, dict):
yield content
elif isinstance(content, str):
yield content
else:
yield from iter([content, ])
else:
return
def __init__(self, name, obj):
super().__init__()
if not name and issubclass(obj.__class__, DOMElement):
name = obj._name
self.name = name
self.obj = obj
if isinstance(obj, GeneratorType):
self._iterable = True
elif issubclass(obj.__class__, DOMElement) or isinstance(obj, str):
self._iterable = False
else:
try:
_ = (e for e in obj)
self._iterable = True
except TypeError:
self._iterable = False
def yield_domgroups(items, kwitems, reverse=False):
"""Flattens the given items/kwitems.
Yields index and DOMGroup after flattening and a DOMGroup.
"reverse" parameter inverts the flattened yielding.
"""
verse = (1, -1)[reverse]
if isinstance(items, GeneratorType):
items = list(items)
unnamed = (DOMGroup(None, item) for item in items[::verse])
named = (DOMGroup(k, v) for k, v in list(kwitems.items())[::verse])
contents = (unnamed, named)[::verse]
for i, group in enumerate(chain(*contents)):
if isinstance(group.obj, DOMElement):
# Is the DOMGroup is a single DOMElement and we have a name we set his name accordingly
group.obj._name = group.name
yield i, group
def do(func, func_args, func_kargs, Monad):
@handle_monadic_throws(Monad)
def run_maybe_iterator():
itr = func(*func_args, **func_kargs)
if isinstance(itr, types.GeneratorType):
@handle_monadic_throws(Monad)
def send(*vals):
try:
# here's the real magic
monad = itr.send(*vals)
return monad.bind(send)
except StopIteration:
return Monad.unit(None)
return send(None)
else:
#not really a generator
if itr is None:
return Monad.unit(None)
else:
return itr
run_maybe_iterator.__name__ = func.__name__
return run_maybe_iterator()
def _find_generators(self, item):
""" A recursive function to flatten generators into lists """
try:
result = []
# Make sure dicts aren't flattened to lists
if isinstance(item, dict):
result = {}
for i in item:
result[self._find_generators(i)] = self._find_generators(item[i])
return result
# Since NoneObjects and strings are both iterable, treat them specially
if isinstance(item, obj.NoneObject) or isinstance(item, str):
return item
if isinstance(item, types.GeneratorType):
raise CacheContainsGenerator
for x in iter(item):
flat_x = self._find_generators(x)
result.append(flat_x)
return result
except TypeError:
return item
def isgenerator(object):
"""Return true if the object is a generator.
Generator objects provide these attributes:
__iter__ defined to support iteration over container
close raises a new GeneratorExit exception inside the
generator to terminate the iteration
gi_code code object
gi_frame frame object or possibly None once the generator has
been exhausted
gi_running set to 1 when generator is executing, 0 otherwise
next return the next item from the container
send resumes the generator and "sends" a value that becomes
the result of the current yield-expression
throw used to raise an exception inside the generator"""
return isinstance(object, types.GeneratorType)
def isgenerator(object):
"""Return true if the object is a generator.
Generator objects provide these attributes:
__iter__ defined to support iteration over container
close raises a new GeneratorExit exception inside the
generator to terminate the iteration
gi_code code object
gi_frame frame object or possibly None once the generator has
been exhausted
gi_running set to 1 when generator is executing, 0 otherwise
next return the next item from the container
send resumes the generator and "sends" a value that becomes
the result of the current yield-expression
throw used to raise an exception inside the generator"""
return isinstance(object, types.GeneratorType)
def test_sort_school(self):
students = [
(3, ("Kyle",)),
(4, ("Christopher", "Jennifer",)),
(6, ("Kareem",))
]
for grade, students_in_grade in students:
for student in students_in_grade:
self.school.add(student, grade)
result = self.school.sort()
# Attempts to catch false positives
self.assertTrue(isinstance(result, Sequence) or
isinstance(result, GeneratorType) or
callable(getattr(result, '__reversed__', False)))
result_list = list(result.items() if hasattr(result, "items")
else result)
self.assertEqual(result_list, students)
def test_iter_files_simple():
gen = utils.iter_files(TEST_DIR, ['py'])
assert isinstance(gen, types.GeneratorType)
assert len(list(gen)) == 1
def new_instance(type, frum, schema=None):
"""
Factory!
"""
if not type2container:
_delayed_imports()
if isinstance(frum, Container):
return frum
elif isinstance(frum, _Cube):
return frum
elif isinstance(frum, _Query):
return _run(frum)
elif isinstance(frum, (list, set, GeneratorType)):
return _ListContainer(frum)
elif isinstance(frum, basestring):
# USE DEFAULT STORAGE TO FIND Container
if not config.default.settings:
Log.error("expecting jx_python.query.config.default.settings to contain default elasticsearch connection info")
settings = set_default(
{
"index": join_field(split_field(frum)[:1:]),
"name": frum,
},
config.default.settings
)
settings.type = None # WE DO NOT WANT TO INFLUENCE THE TYPE BECAUSE NONE IS IN THE frum STRING ANYWAY
return type2container["elasticsearch"](settings)
elif isinstance(frum, Mapping):
frum = wrap(frum)
if frum.type and type2container[frum.type]:
return type2container[frum.type](frum.settings)
elif frum["from"]:
frum = copy(frum)
frum["from"] = Container(frum["from"])
return _Query.wrap(frum)
else:
Log.error("Do not know how to handle {{frum|json}}", frum=frum)
else:
Log.error("Do not know how to handle {{type}}", type=frum.__class__.__name__)
def expand(item):
if isinstance(item, types.GeneratorType):
return list(item)
if isinstance(item, dict):
return ['{0}={1}'.format(k, v) for k, v in item.items()]
return item
