def create_session(self, loop):
        conn = None

        if self.proxy and self.proxy_user:
            conn = aiohttp.ProxyConnector(
                loop=loop,
                limit=self.parallel,
                proxy=self.proxy,
                proxy_auth=aiohttp.BasicAuth(self.proxy_user, self.proxy_password)
            )
        elif self.proxy:
            conn = aiohttp.ProxyConnector(loop=loop, limit=self.parallel, proxy=self.proxy)
        else:
            conn = aiohttp.TCPConnector(loop=loop, limit=self.parallel)

        session = aiohttp.ClientSession(connector=conn)
        return session 

def listen_message_stream(self, id_blacklist=None):
        id_blacklist = set(id_blacklist or [self.me, ])

        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        with aiohttp.ClientSession(loop=loop) as session:
            self.aioclient_session = session

            tasks = [
                asyncio.ensure_future(self.fetch(session, room, id_blacklist))
                for room in self.rooms
            ]
            done, _ = loop.run_until_complete(
                asyncio.wait(tasks, return_when=asyncio.FIRST_EXCEPTION)
            )
            for d in done:
                if d.exception():
                    raise d.exception() 

def get_launch_dict(launch_number: int = 0) -> Dict:
    """Get a launch information dictionary for the given launch.

    If launch_number <= 0 (the default), get the "next" launch.
    """

    route = launch_number if launch_number > 0 else "next"
    spacex_api_url = f"https://api.spacexdata.com/v3/launches/{route}"

    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(spacex_api_url) as response:
                if response.status != 200:
                    logging.error(f"Response status: {response.status}")
                    return {}
                return await response.json()

    except aiohttp.client_exceptions.ClientConnectorError:
        logging.error("Cannot connect to api.spacexdata.com")
        return {}

    except aiohttp.ContentTypeError:
        logging.error("JSON decode failed")
        return {} 

def test_returns_answer_with_timeout(self):
        question: str = "how do I clean the stove?"
        options = QnAMakerOptions(timeout=999999)
        qna = QnAMaker(QnaApplicationTest.tests_endpoint, options)
        context = QnaApplicationTest._get_context(question, TestAdapter())
        response_json = QnaApplicationTest._get_json_for_file("ReturnsAnswer.json")

        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            result = await qna.get_answers(context, options)

            self.assertIsNotNone(result)
            self.assertEqual(
                options.timeout, qna._generate_answer_helper.options.timeout
            ) 

def test_should_filter_low_score_variation(self):
        options = QnAMakerOptions(top=5)
        qna = QnAMaker(QnaApplicationTest.tests_endpoint, options)
        question: str = "Q11"
        context = QnaApplicationTest._get_context(question, TestAdapter())
        response_json = QnaApplicationTest._get_json_for_file("TopNAnswer.json")

        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            results = await qna.get_answers(context)
            self.assertEqual(4, len(results), "Should have received 4 answers.")

            filtered_results = qna.get_low_score_variation(results)
            self.assertEqual(
                3,
                len(filtered_results),
                "Should have 3 filtered answers after low score variation.",
            ) 

def test_should_answer_with_prompts(self):
        options = QnAMakerOptions(top=2)
        qna = QnAMaker(QnaApplicationTest.tests_endpoint, options)
        question: str = "how do I clean the stove?"
        turn_context = QnaApplicationTest._get_context(question, TestAdapter())
        response_json = QnaApplicationTest._get_json_for_file("AnswerWithPrompts.json")

        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            results = await qna.get_answers(turn_context, options)
            self.assertEqual(1, len(results), "Should have received 1 answers.")
            self.assertEqual(
                1, len(results[0].context.prompts), "Should have received 1 prompt."
            ) 

def test_should_answer_with_high_score_provided_context(self):
        qna = QnAMaker(QnaApplicationTest.tests_endpoint)
        question: str = "where can I buy?"
        context = QnARequestContext(
            previous_qna_id=5, previous_user_query="how do I clean the stove?"
        )
        options = QnAMakerOptions(top=2, qna_id=55, context=context)
        turn_context = QnaApplicationTest._get_context(question, TestAdapter())
        response_json = QnaApplicationTest._get_json_for_file(
            "AnswerWithHighScoreProvidedContext.json"
        )

        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            results = await qna.get_answers(turn_context, options)
            self.assertEqual(1, len(results), "Should have received 1 answers.")
            self.assertEqual(1, results[0].score, "Score should be high.") 

def test_should_answer_with_low_score_without_provided_context(self):
        qna = QnAMaker(QnaApplicationTest.tests_endpoint)
        question: str = "where can I buy?"
        options = QnAMakerOptions(top=2, context=None)

        turn_context = QnaApplicationTest._get_context(question, TestAdapter())
        response_json = QnaApplicationTest._get_json_for_file(
            "AnswerWithLowScoreProvidedWithoutContext.json"
        )

        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            results = await qna.get_answers(turn_context, options)
            self.assertEqual(
                2, len(results), "Should have received more than one answers."
            )
            self.assertEqual(True, results[0].score < 1, "Score should be low.") 

def _get_service_result(
        cls,
        utterance: str,
        response_file: str,
        bot_adapter: BotAdapter = TestAdapter(),
        options: QnAMakerOptions = None,
    ) -> [dict]:
        response_json = QnaApplicationTest._get_json_for_file(response_file)

        qna = QnAMaker(QnaApplicationTest.tests_endpoint)
        context = QnaApplicationTest._get_context(utterance, bot_adapter)

        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            result = await qna.get_answers(context, options)

            return result 

def _get_service_result_raw(
        cls,
        utterance: str,
        response_file: str,
        bot_adapter: BotAdapter = TestAdapter(),
        options: QnAMakerOptions = None,
    ) -> [dict]:
        response_json = QnaApplicationTest._get_json_for_file(response_file)

        qna = QnAMaker(QnaApplicationTest.tests_endpoint)
        context = QnaApplicationTest._get_context(utterance, bot_adapter)

        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            result = await qna.get_answers_raw(context, options)

            return result 

def __init__(self, token, session=None, is_async=False, **options):
        self.token = token
        self.is_async = is_async
        self.error_debug = options.get('error_debug', False)
        self.timeout = options.get('timeout', 10)
        self.api = API(options.get('url', 'https://api.clashroyale.com/v1'))
        self.session = session or (aiohttp.ClientSession() if is_async else requests.Session())
        self.camel_case = options.get('camel_case', False)
        self.headers = {
            'Authorization': 'Bearer {}'.format(token),
            'User-Agent': 'python-clashroyale-client (fourjr/kyb3r) ' + options.get('user_agent', '')
        }
        self.cache_fp = options.get('cache_fp')
        self.using_cache = bool(self.cache_fp)
        self.cache_reset = options.get('cache_expires', 300)
        if self.using_cache:
            table = options.get('table_name', 'cache')
            self.cache = SqliteDict(self.cache_fp, table)

        constants = options.get('constants')
        if not constants:
            with Path(__file__).parent.parent.joinpath('constants.json').open(encoding='utf8') as f:
                constants = json.load(f)
        self.constants = BaseAttrDict(self, constants, None) 

def __init__(self, token, session=None, is_async=False, **options):
        self.token = token
        self.is_async = is_async
        self.error_debug = options.get('error_debug', False)
        self.timeout = options.get('timeout', 10)
        self.api = API(options.get('url', 'https://api.royaleapi.com'))
        self.session = session or (aiohttp.ClientSession() if is_async else requests.Session())
        self.camel_case = options.get('camel_case', False)
        self.headers = {
            'Authorization': 'Bearer {}'.format(token),
            'User-Agent': 'python-clashroyale-client (fourjr/kyb3r) ' + options.get('user_agent', '')
        }
        self.cache_fp = options.get('cache_fp')
        self.using_cache = bool(self.cache_fp)
        self.cache_reset = options.get('cache_expires', 300)
        self.ratelimit = [10, 10, 0]
        if self.using_cache:
            table = options.get('table_name', 'cache')
            self.cache = SqliteDict(self.cache_fp, table) 

def test_http_session(self, loop, local_timeout_server, consul_port):
        async def test_session_close():
            http_server = await local_timeout_server
            c = consul.aio.Consul(port=http_server.port, loop=loop)
            c.agent.services()

            c.http._session = aiohttp.ClientSession()

            assert not c.http._session.closed
            c.http.__del__()
            await c.http.close()
            assert c.http._session.closed
            http_server.server.stop()
            ...

        loop.run_until_complete(test_session_close()) 

def _request(self, callback, method, uri, data=None, headers=None):
        connector = aiohttp.TCPConnector(loop=self._loop,
                                         verify_ssl=self.verify)
        async with aiohttp.ClientSession(connector=connector) as session:
            self._session = session
            resp = await session.request(method=method,
                                         url=uri,
                                         data=data,
                                         headers=headers)
            body = await resp.text(encoding='utf-8')
            content = await resp.read()
            if resp.status == 599:
                raise base.Timeout
            r = base.Response(resp.status, resp.headers, body, content)
            await session.close()
            return callback(r)

    # python prior 3.4.1 does not play nice with __del__ method 

def test_create(self, mocker):
        # Setup
        import asyncio

        session_cls_mock = mocker.patch("aiohttp.ClientSession")
        positionals = [1]
        keywords = {"keyword": 2}

        # Run: Create client
        client = aiohttp_.AiohttpClient.create(*positionals, **keywords)

        # Verify: session hasn't been created yet.
        assert not session_cls_mock.called

        # Run: Get session
        loop = asyncio.get_event_loop()
        loop.run_until_complete(asyncio.ensure_future(client.session()))

        # Verify: session created with args
        session_cls_mock.assert_called_with(*positionals, **keywords) 

def test_close_auto_created_session(self, mocker):
        # Setup
        import asyncio
        import gc
        import aiohttp

        mock_session = mocker.Mock(spec=aiohttp.ClientSession)
        session_cls_mock = mocker.patch("aiohttp.ClientSession")
        session_cls_mock.return_value = mock_session

        positionals = [1]
        keywords = {"keyword": 2}

        # Run: Create client
        client = aiohttp_.AiohttpClient.create(*positionals, **keywords)

        # Run: Get session
        loop = asyncio.get_event_loop()
        loop.run_until_complete(asyncio.ensure_future(client.session()))

        # Verify: session created with args
        session_cls_mock.assert_called_with(*positionals, **keywords)
        del client
        gc.collect()
        session_cls_mock.return_value.close.assert_called_with() 

def create(cls, *args, **kwargs):
        """
        Builds a client instance with
        :py:class:`aiohttp.ClientSession` arguments.

        Instead of directly initializing this class with a
        :py:class:`aiohttp.ClientSession`, use this method to have the
        client lazily construct a session when sending the first
        request. Hence, this method guarantees that the creation of the
        underlying session happens inside of a coroutine.

        Args:
            *args: positional arguments that
                :py:class:`aiohttp.ClientSession` takes.
            **kwargs: keyword arguments that
                :py:class:`aiohttp.ClientSession` takes.
        """
        session_build_args = cls._create_session(*args, **kwargs)
        return AiohttpClient(session=session_build_args) 

def get_http_client(app):
    """ get http client """
    if "client" in app:
        return app["client"]

    # first time call, create client interface
    # use shared client so that all client requests
    #   will share the same connection pool
    if "loop" not in app:
        raise KeyError("loop not initialized")
    loop = app["loop"]
    max_tcp_connections = int(config.get("max_tcp_connections"))
    log.info(f"Initiating TCPConnector with limit {max_tcp_connections} connections")
    client = ClientSession(loop=loop, connector=TCPConnector(limit_per_host=max_tcp_connections))
    #create the app object
    app['client'] = client
    return client 

def make_request(
        self,
        url: str,
        session: ClientSession,
        **params
    ):
        assert isinstance(session, ClientSession)

        for key in list(params.keys()):
            if params[key] is None:
                del params[key]
            elif isinstance(params[key], list):
                params[key] = ",".join(params[key])

        async with session.get(
            url, params=params, headers=self.headers
        ) as resp:
            response = await resp.json()
            if "error" in response:
                raise DarkSkyException(response["code"], response["error"])
        response["timezone"] = params.get("timezone") or response["timezone"]
        return response 

def get_forecast(
        self,
        latitude: float,
        longitude: float,
        client_session: aiohttp.ClientSession,
        extend: bool = None,
        lang=languages.ENGLISH,
        values_units=units.AUTO,
        exclude: [weather] = None,
        timezone: str = None,
    ) -> Forecast:
        url = self.get_url(latitude, longitude)
        data = await self.request_manager.make_request(
            url=url,
            extend=weather.HOURLY if extend else None,
            lang=lang,
            units=values_units,
            exclude=exclude,
            timezone=timezone,
            session=client_session,
        )
        return Forecast(**data) 

def get_time_machine_forecast(
        self,
        latitude: float,
        longitude: float,
        time: datetime,
        client_session: aiohttp.ClientSession,
        extend: bool = False,
        lang=languages.ENGLISH,
        values_units=units.AUTO,
        exclude: [weather] = None,
        timezone: str = None
    ) -> Forecast:
        url = self.get_url(latitude, longitude, int(time.timestamp()))
        data = await self.request_manager.make_request(
            url=url,
            extend=weather.HOURLY if extend else None,
            lang=lang,
            units=values_units,
            exclude=exclude,
            timezone=timezone,
            session=client_session,
        )
        return Forecast(**data) 

def get_forecast_async():
    async def get_async_data():
        darksky = DarkSkyAsync("api_key")
        with aioresponses.aioresponses() as resp:
            resp.get(re.compile(".+"), status=200, payload=copy.deepcopy(DATA))

            result = await darksky.get_forecast(
                DATA["latitude"],
                DATA["longitude"],
                client_session=aiohttp.ClientSession()
            )

        return result

    loop = asyncio.get_event_loop()
    return loop.run_until_complete(get_async_data()) 

def view_logs(server: str, token: str) -> None:
    async with ClientSession() as session:
        async with session.ws_connect(f"{server}/_matrix/maubot/v1/logs") as ws:
            await ws.send_str(token)
            try:
                msg: WSMessage
                async for msg in ws:
                    if msg.type == WSMsgType.TEXT:
                        if not handle_msg(msg.json()):
                            break
                    elif msg.type == WSMsgType.ERROR:
                        print(Fore.YELLOW + "Connection error: " + msg.data + Fore.RESET)
                    elif msg.type == WSMsgType.CLOSE:
                        print(Fore.YELLOW + "Server closed connection" + Fore.RESET)
            except asyncio.CancelledError:
                pass 

def __init__(
        self,
        url: str,
        headers: Optional[LooseHeaders] = None,
        cookies: Optional[LooseCookies] = None,
        auth: Optional[BasicAuth] = None,
        ssl: Union[SSLContext, bool, Fingerprint] = False,
        timeout: Optional[int] = None,
        client_session_args: Dict[str, Any] = {},
    ) -> None:
        """Initialize the transport with the given aiohttp parameters.

        :param url: The GraphQL server URL. Example: 'https://server.com:PORT/path'.
        :param headers: Dict of HTTP Headers.
        :param cookies: Dict of HTTP cookies.
        :param auth: BasicAuth object to enable Basic HTTP auth if needed
        :param ssl: ssl_context of the connection. Use ssl=False to disable encryption
        :param client_session_args: Dict of extra args passed to aiohttp.ClientSession
        """
        self.url: str = url
        self.headers: Optional[LooseHeaders] = headers
        self.cookies: Optional[LooseCookies] = cookies
        self.auth: Optional[BasicAuth] = auth
        self.ssl: Union[SSLContext, bool, Fingerprint] = ssl
        self.timeout: Optional[int] = timeout
        self.client_session_args = client_session_args

        self.session: Optional[aiohttp.ClientSession] = None 

def connect(self) -> None:
        """Coroutine which will:

        - create an aiohttp ClientSession() as self.session

        Should be cleaned with a call to the close coroutine
        """

        if self.session is None:

            client_session_args: Dict[str, Any] = {
                "cookies": self.cookies,
                "headers": self.headers,
                "auth": self.auth,
            }

            if self.timeout is not None:
                client_session_args["timeout"] = aiohttp.ClientTimeout(
                    total=self.timeout
                )

            # Adding custom parameters passed from init
            client_session_args.update(self.client_session_args)

            self.session = aiohttp.ClientSession(**client_session_args)

        else:
            raise TransportAlreadyConnected("Transport is already connected") 

def main(loop):
    async with aiohttp.ClientSession(loop=loop) as session:
        client = AiohttpClient(session, "http://localhost:5000")
        response = await client.request("ping")
    print(response.data.result) 

def main(loop):

    async with aiohttp.ClientSession(loop=loop) as session:
        client = AiohttpClient(session, "http://localhost:5000")
        requests = [Request("ping"), Notification("ping"), Request("ping")]
        response = await client.send(requests)

    for data in response.data:
        if data.ok:
            print("{}: {}".format(data.id, data.result))
        else:
            logging.error("%d: %s", data.id, data.message) 

def dl(self,url,name):
        name=removeDisallowedFilenameChars(str(name))
        with open(name,"wb+") as fid:
            async with aiohttp.ClientSession() as session:
                async with session.get(url) as stream:
                    chunk_size=1024*32
                    size=int(stream.headers.get("Content-Length",1))
                    while True:
                        chunk = await stream.content.read(chunk_size)
                        if not chunk: break
                        fid.write(chunk)                    
                        self.emit("percent",int(( fid.tell()/size)*100))
        return name 

def request(
    url, data=None, headers={}
):  # mimic urllib.Request() (GET & POST only)
    async with aiohttp.ClientSession(cookie_jar=jar) as session:
        try:
            if data:
                async with session.post(
                    url, data=data, headers=headers, ssl=False
                ) as resp:
                    return Response(
                        resp.status, await resp.text(), headers=resp.headers
                    )
            else:
                async with session.get(url, headers=headers, ssl=False) as resp:
                    return Response(
                        resp.status, await resp.text(), headers=resp.headers
                    )
        except aiohttp.client_exceptions.ClientConnectorError as e:
            return Response(None, str(e))


# -=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=


# Async aiohttp things (use current)
############################################################# 

def _post_count_to_bot_list(bl_data: Dict[str, str], guild_count: int) -> None:
    async with aiohttp.ClientSession() as session:
        await session.post(
            bl_data["url"],
            json={bl_data["guild_count_parameter"]: guild_count},
            headers={
                "Authorization": bl_data["token"],
                "Content-Type": "application/json",
            },
        )