def subscribe(
        self, document: DocumentNode, *args, **kwargs
    ) -> Generator[Dict, None, None]:
        """Execute a GraphQL subscription with a python generator.

        We need an async transport for this functionality.
        """

        async_generator = self.subscribe_async(document, *args, **kwargs)

        loop = asyncio.get_event_loop()

        assert not loop.is_running(), (
            "Cannot run client.subscribe if an asyncio loop is running."
            " Use subscribe_async instead."
        )

        try:
            while True:
                result = loop.run_until_complete(async_generator.__anext__())
                yield result

        except StopAsyncIteration:
            pass 

def gather(self, wnid, relative="", include_subset=False):
        loop = asyncio.get_event_loop()
        session = self.create_session(loop)
        folders = []

        f = loop.run_until_complete(self.download_images(session, wnid, relative))
        folders.append(f)

        if include_subset:
            wnids = self._get_subsets(wnid)
            path = self.file_api.join_relative(relative, f)
            downloads = asyncio.wait([self.download_images(session, wnid, path) for wnid in wnids])
            done, pending = loop.run_until_complete(downloads)
            folders += [d.result() for d in done]

        session.close()

        return folders 

def test():
    class aeff(wuy.Window):
        size = (100, 100)

        def init(self):
            asyncio.get_event_loop().call_later(2, self.exit)

    # !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
    # the following line is needed
    # because pytest seems to execute from a different path
    # then the executable one (think freezed)
    # ex: it works without it, in a real context
    # ex: it's needed when pytest execute the test
    # IRL : it's not needed to change the path
    # !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
    wuy.PATH = os.getcwd()  # !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
    # !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!

    cf = "web/aeff.html"
    if os.path.isfile(cf):
        os.unlink(cf)
    aeff()
    assert os.path.isfile(cf), "a default file can't be created !!!"
    os.unlink(cf) 

def async_run(tasks):
    """
    run a group of tasks async
    Requires the tasks arg to be a list of functools.partial()
    """
    if not tasks:
        return

    # start a new async event loop
    loop = asyncio.get_event_loop()
    # https://github.com/python/asyncio/issues/258
    executor = concurrent.futures.ThreadPoolExecutor(5)
    loop.set_default_executor(executor)

    async_tasks = [asyncio.ensure_future(async_task(task, loop)) for task in tasks]
    # run tasks in parallel
    loop.run_until_complete(asyncio.wait(async_tasks))
    # deal with errors (exceptions, etc)
    for task in async_tasks:
        error = task.exception()
        if error is not None:
            raise error

    executor.shutdown(wait=True) 

def main(_):
  urls = get_urls_for_shard_group(
      FLAGS.urls_dir, FLAGS.shard_id, FLAGS.group_id)
  tf.logging.info("Fetching %d URLs for shard %d, group %d",
                  len(urls), FLAGS.shard_id, FLAGS.group_id)

  tf.gfile.MakeDirs(FLAGS.out_dir)
  out_fname = tfrecord_fname(FLAGS.out_dir, FLAGS.shard_id)

  with utils.timing("group_fetch"):
    logging_fnames = {}
    if FLAGS.log_samples:
      logging_fnames["samples"] = os.path.join(
          FLAGS.out_dir, "samples.%d.txt" % FLAGS.shard_id)
    loop = asyncio.get_event_loop()
    num_written = loop.run_until_complete(asyncio.ensure_future(
        fetch_urls(urls,
                   out_fname,
                   logging_fnames)))

  tf.logging.info("Total URLs: %d", len(urls))
  tf.logging.info("Num written: %d", num_written)
  tf.logging.info("Coverage: %.1f", (num_written / len(urls)) * 100) 

def start_work():
    global aredis
    loop = asyncio.get_event_loop()
    aredis = await aioredis.create_redis('redis://localhost', loop=loop)

    if await aredis.get('state') == b'running':
        return "<center>Please wait for current work to finish.</center>"
    else:
        await aredis.set('state', 'ready')

    if await aredis.get('state') == b'ready':
        loop.create_task(some_work())
        body = '''
        <center>
        work started!
        </center>
        <script type="text/javascript">
            window.location = "''' + url_for('progress') + '''";
        </script>'''
        return body 

def sync_with_context(future: Awaitable) -> Any:
    context = None
    if _request_ctx_stack.top is not None:
        context = _request_ctx_stack.top.copy()
    elif _websocket_ctx_stack.top is not None:
        context = _websocket_ctx_stack.top.copy()
    elif _app_ctx_stack.top is not None:
        context = _app_ctx_stack.top.copy()

    async def context_wrapper() -> Any:
        if context is not None:
            async with context:
                return await future
        else:
            return await future

    return asyncio.get_event_loop().sync_wait(context_wrapper())  # type: ignore 

def __anext__(self):
        loop = asyncio.get_event_loop()
        try:
            chunk = await loop.run_in_executor(
                None, _msrest_next, self.iter_content_func
            )
            if not chunk:
                raise _MsrestStopIteration()
            if self.user_callback and callable(self.user_callback):
                self.user_callback(chunk, self.response)
            return chunk
        except _MsrestStopIteration:
            self.response.close()
            raise StopAsyncIteration()
        except Exception as err:
            _LOGGER.warning("Unable to stream download: %s", err)
            self.response.close()
            raise 

def main():
    loop = asyncio.get_event_loop()
    big_brother = None
    try:
        pid_path = os.path.join(app_dir.user_cache_dir, 'trader.pid')
        if not os.path.exists(pid_path):
            if not os.path.exists(app_dir.user_cache_dir):
                os.makedirs(app_dir.user_cache_dir)
        with open(pid_path, 'w') as pid_file:
            pid_file.write(str(os.getpid()))
        big_brother = TradeStrategy(io_loop=loop)
        print('Big Brother is watching you!')
        print('used config file:', config_file)
        print('log stored in:', app_dir.user_log_dir)
        print('pid file:', pid_path)
        loop.create_task(big_brother.install())
        loop.run_forever()
    except KeyboardInterrupt:
        pass
    except Exception as ee:
        logger.info('发生错误: %s', repr(ee), exc_info=True)
    finally:
        big_brother and loop.run_until_complete(big_brother.uninstall())
        logger.info('程序已退出') 

def fetch_bar2():
    day = datetime.datetime.strptime('20160114', '%Y%m%d').replace(tzinfo=pytz.FixedOffset(480))
    end = datetime.datetime.strptime('20160914', '%Y%m%d').replace(tzinfo=pytz.FixedOffset(480))
    while day <= end:
        day, trading = await is_trading_day(day)
        if trading:
            print('process ', day)
            tasks = [
                asyncio.ensure_future(update_from_shfe(day)),
                asyncio.ensure_future(update_from_dce(day)),
                asyncio.ensure_future(update_from_czce(day)),
                asyncio.ensure_future(update_from_cffex(day)),
            ]
            await asyncio.wait(tasks)
        day += datetime.timedelta(days=1)
    print('all done!')


# asyncio.get_event_loop().run_until_complete(fetch_bar2())
# create_main_all()
# fetch_from_quandl_all()
# clean_dailybar()
# load_kt_data() 

def __init__(self, io_loop: asyncio.AbstractEventLoop = None):
        super().__init__()
        self.io_loop = io_loop or asyncio.get_event_loop()
        self.sub_client = self.io_loop.run_until_complete(
                aioredis.create_redis((config.get('REDIS', 'host', fallback='localhost'),
                                       config.getint('REDIS', 'port', fallback=6379)),
                                      db=config.getint('REDIS', 'db', fallback=1)))
        self.redis_client = redis.StrictRedis(
            host=config.get('REDIS', 'host', fallback='localhost'),
            db=config.getint('REDIS', 'db', fallback=1), decode_responses=True)
        self.initialized = False
        self.sub_tasks = list()
        self.sub_channels = list()
        self.channel_router = dict()
        self.crontab_router = defaultdict(dict)
        self.datetime = None
        self.time = None
        self.loop_time = None 

def db_dsn_fixture(request):
    loop = asyncio.get_event_loop()

    pg_dsn = 'postgresql://postgres:postgres@postgres:5432/postgres'
    db_name = loop.run_until_complete(init_db(pg_dsn, loop=loop))

    db_dsn = 'postgresql://postgres:postgres@postgres:5432/{}'.format(db_name)
    loop.run_until_complete(setup_db(db_dsn, loop=loop))

    def fin():
        loop.run_until_complete(drop_db(pg_dsn, db_name, loop=loop))

    request.addfinalizer(fin)
    return db_dsn

# define graph 

def execute(self, document: DocumentNode, *args, **kwargs) -> Dict:
        """Execute the provided document AST against the configured remote server.

        This function WILL BLOCK until the result is received from the server.

        Either the transport is sync and we execute the query synchronously directly
        OR the transport is async and we execute the query in the asyncio loop
        (blocking here until answer).
        """

        if isinstance(self.transport, AsyncTransport):

            loop = asyncio.get_event_loop()

            assert not loop.is_running(), (
                "Cannot run client.execute if an asyncio loop is running."
                " Use execute_async instead."
            )

            data: Dict[Any, Any] = loop.run_until_complete(
                self.execute_async(document, *args, **kwargs)
            )

            return data

        else:  # Sync transports
            return self.execute_sync(document, *args, **kwargs) 

def main():
  print("Creating our event loop")
  loop = asyncio.get_event_loop()
  
  loop.run_forever()
  print("Our Loop will now run forever, this will never execute") 

def main():
  loop = asyncio.get_event_loop()
  loop.run_until_complete(myWork())
  loop.stop()
  print("Loop Stopped")
  loop.close()

  print(loop.is_closed()) 

def main():
  loop = asyncio.get_event_loop()
  task = loop.create_task(slow_operation())
  task.add_done_callback(got_result)
  loop.run_until_complete(task) 

def main():
   loop = asyncio.get_event_loop()
   loop.run_until_complete(myGenerator())
   loop.close() 

def generate_batches(self, size):
        mean = self.__load_mean()

        async def to_array(im):
            im.load()
            converted = self.convert(im)
            arr = self.__to_array(converted, mean)
            im.image = None  # don't use image any more, so release reference
            return arr, im.label

        batch = []
        loop = asyncio.get_event_loop()

        for im in self.label_file.fetch(load_image=False):
            batch.append(to_array(im))

            if len(batch) == size:
                tasks = asyncio.wait(batch)
                done, pending = loop.run_until_complete(tasks)
                results = []
                for d in done:
                    try:
                        results.append(d.result())
                    except:
                        pass

                x_sample, y_sample = results[0]
                x_batch = np.ndarray((size,) + x_sample.shape, x_sample.dtype)
                y_batch = np.ndarray((size,), np.int32)

                for j, r in enumerate(results):
                    x_batch[j], y_batch[j] = r

                yield x_batch, y_batch
                i = 0
                batch.clear() 

def __init__(
        self, config: Config, registry: MetricsRegistry, logger: Logger,
    ):
        self._config = config
        self._registry = registry
        self._logger = logger
        self._timed_queries: List[Query] = []
        self._aperiodic_queries: List[Query] = []
        # map query names to their TimedCalls
        self._timed_calls: Dict[str, TimedCall] = {}
        # map query names to list of database names
        self._doomed_queries: Dict[str, Set[str]] = defaultdict(set)
        self._loop = asyncio.get_event_loop()
        self._setup() 

def run_sync(func, *args, **kwargs):
    """Run a non-async function in a new thread and return an awaitable"""
    # Returning a coro
    return asyncio.get_event_loop().run_in_executor(None, functools.partial(func, *args, **kwargs)) 

def _team_name(self, server, team):
        team = self._safe_path(team).lower()
        return self._format_name(self.teams[server.id]["TEAMS"][team]["NAME"])




    # try:
    #     # of course edit once in bot
    #     # don't forget to correct paths
    #     loop = asyncio.get_event_loop()
    #     loop.run_until_complete(loop())
    # except EOFError:
    #     pass 

def ready_up(self):
        self.loop = asyncio.get_event_loop() 

def __init__(self, *args, **kwargs):
        self._loop = kwargs['loop'] if 'loop' in kwargs \
            else asyncio.get_event_loop()
        self._is_running = False
        self._run_complete = asyncio.Future(loop = self._loop) 

def _exit(instance=None):  # exit method
    global application

    if asyncio.get_event_loop().is_running():
        asyncio.get_event_loop().stop()

    if instance and hasattr(instance, "_browser") and instance._browser:
        del instance._browser
        instance._browser = None

    application = None
    wlog("exit") 

def test():
    class aeff(wuy.Window):
        "test double open"
        size = (100, 100)

        def init(self):
            asyncio.get_event_loop().call_later(2, self.exit)

    aeff()
    aeff() 

def test():
    class saeff1(wuy.Server):
        "I'm a server"

        def init(self):
            asyncio.get_event_loop().call_later(2, self.exit)

    class saeff2(wuy.Server):
        "I'm a server, and I will killed by saeff1"
        pass

    wuy.Server.run()
    assert "saeff1" in wuy.currents
    assert "saeff2" in wuy.currents 

def test_a_server():
    class saeff(wuy.Server):
        "I'm a server"

        def init(self):
            asyncio.get_event_loop().call_later(2, self.exit)

    saeff()
    assert "saeff" in wuy.currents 

def init(self):             #<- special method which is called at the start !
        self.emit("setTheDate",datetime.datetime.now())
        asyncio.get_event_loop().call_later(1, self.init) 

def _get_task(self):
        try:
            # Prevent failure when run from a thread
            # without an event loop.
            loop = asyncio.get_event_loop()
        except RuntimeError:
            return None

        return asyncio.Task.current_task(loop=loop) 

def run_test(self, test_fn):
        @asyncio.coroutine
        def async_test_fn():
            test_fn()
        asyncio.get_event_loop().run_until_complete(async_test_fn())