Aiohttp2 support+ssl fix

.gitignore

@@ -1,5 +1,6 @@
 # python specific
 env*
+.cache/
 *.pyc
 *.so
 *.pyd
@@ -9,6 +10,7 @@ MANIFEST
 __pycache__/
 *.egg-info/
 .coverage
+.python-version
 htmlcov
 
 # generic files to ignore

.travis.yml

@@ -13,6 +13,7 @@ env:
   - ES="5.0.2"
   - ES="5.1.2"
   - ES="5.2.2"
+  - ES="5.3.0"
 install:
   - curl -O https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-$ES.deb && sudo dpkg -i --force-confnew elasticsearch-$ES.deb && sudo service elasticsearch start
   - pip install -U setuptools

aioelasticsearch/__init__.py

@@ -43,6 +43,5 @@ def close(self):
         def __aenter__(self):  # noqa
             return self
 
-        @asyncio.coroutine
         def __aexit__(self, *exc_info):  # noqa
-            yield from self.close()
+            return self.close()

aioelasticsearch/compat.py

@@ -2,10 +2,14 @@
 import sys
 from functools import partial
 
+import aiohttp
+
 PY_344 = sys.version_info >= (3, 4, 4)
 PY_350 = sys.version_info >= (3, 5, 0)
 PY_352 = sys.version_info >= (3, 5, 2)
 
+AIOHTTP_2 = aiohttp.__version__ >= '2.0.0'
+
 
 def create_task(*, loop=None):
     if loop is None:

aioelasticsearch/connection.py

@@ -1,11 +1,19 @@
 import asyncio
+import ssl
 
 import aiohttp
-from aiohttp.errors import ClientError, FingerprintMismatch
-from elasticsearch.connection import Connection
-from elasticsearch.exceptions import (ConnectionError, ConnectionTimeout,
+
+from aioelasticsearch.compat import AIOHTTP_2, create_future
+
+if AIOHTTP_2:
+    from aiohttp import ClientError
+else:
+    from aiohttp.errors import ClientError
+
+from elasticsearch.connection import Connection  # noqa # isort:skip
+from elasticsearch.exceptions import (ConnectionError, ConnectionTimeout,  # noqa # isort:skip
                                       SSLError)
-from yarl import URL
+from yarl import URL  # noqa # isort:skip
 
 
 class AIOHttpConnection(Connection):
@@ -16,6 +24,7 @@ def __init__(
         port=9200,
         http_auth=None,
         use_ssl=False,
+        ssl_context=None,
         verify_certs=False,
         maxsize=10,
         headers=None,
@@ -54,51 +63,118 @@ def __init__(
                 connector=aiohttp.TCPConnector(
                     limit=maxsize,
                     use_dns_cache=kwargs.get('use_dns_cache', False),
+                    ssl_context=ssl_context,
                     verify_ssl=self.verify_certs,
-                    conn_timeout=None,
                     loop=self.loop,
                 ),
             )
 
     def close(self):
-        return self.session.close()
+        coro = self.session.close()
+        if not AIOHTTP_2:
+            return coro
+
+        future = create_future(loop=self.loop)
+        future.set_result(None)
+        return future
 
     @asyncio.coroutine
-    def perform_request(self, method, url, params=None, body=None, timeout=None, ignore=()):  # noqa
+    def perform_request(
+        self,
+        method,
+        url,
+        params=None,
+        body=None,
+        timeout=None,
+        ignore=()
+    ):
         url_path = url
 
         url = (self.base_url / url.lstrip('/')).with_query(params)
 
         start = self.loop.time()
         response = None
         try:
-            with aiohttp.Timeout(timeout or self.timeout, loop=self.loop):  # noqa
-                response = yield from self.session.request(method, url, data=body, headers=self.headers, timeout=None)  # noqa
+            with aiohttp.Timeout(timeout or self.timeout, loop=self.loop):
+                response = yield from self.session.request(
+                    method,
+                    url,
+                    data=body,
+                    headers=self.headers,
+                    timeout=None,
+                )
                 raw_data = yield from response.text()
 
             duration = self.loop.time() - start
 
+        except ssl.CertificateError as exc:
+            self.log_request_fail(
+                method,
+                url,
+                url_path,
+                body,
+                self.loop.time() - start,
+                exception=exc,
+            )
+            raise SSLError('N/A', str(exc), exc)
+
         except asyncio.TimeoutError as exc:
-            self.log_request_fail(method, url, url_path, body, self.loop.time() - start, exception=exc)  # noqa
+            self.log_request_fail(
+                method,
+                url,
+                url_path,
+                body,
+                self.loop.time() - start,
+                exception=exc,
+            )
             raise ConnectionTimeout('TIMEOUT', str(exc), exc)
 
-        except FingerprintMismatch as exc:
-            self.log_request_fail(method, url, url_path, body, self.loop.time() - start, exception=exc)  # noqa
-            raise SSLError('N/A', str(exc), exc)
-
         except ClientError as exc:
-            self.log_request_fail(method, url, url_path, body, self.loop.time() - start, exception=exc)  # noqa
-            raise ConnectionError('N/A', str(exc), exc)
+            self.log_request_fail(
+                method,
+                url,
+                url_path,
+                body,
+                self.loop.time() - start,
+                exception=exc,
+            )
+
+            _exc = str(exc)
+            # aiohttp wraps ssl error
+            if 'SSL: CERTIFICATE_VERIFY_FAILED' in _exc:
+                raise SSLError('N/A', _exc, exc)
+
+            raise ConnectionError('N/A', _exc, exc)
 
         finally:
             if response is not None:
                 yield from response.release()
 
-        # raise errors based on http status codes, let the client handle those if needed  # noqa
-        if not (200 <= response.status < 300) and response.status not in ignore:  # noqa
-            self.log_request_fail(method, url, url_path, body, duration, response.status, raw_data)  # noqa
+        # raise errors based on http status codes
+        # let the client handle those if needed
+        if (
+            not (200 <= response.status < 300) and
+            response.status not in ignore
+        ):
+            self.log_request_fail(
+                method,
+                url,
+                url_path,
+                body,
+                duration,
+                response.status,
+                raw_data,
+            )
             self._raise_error(response.status, raw_data)
 
-        self.log_request_success(method, url, url_path, body, response.status, raw_data, duration)  # noqa
+        self.log_request_success(
+            method,
+            url,
+            url_path,
+            body,
+            response.status,
+            raw_data,
+            duration,
+        )
 
         return response.status, response.headers, raw_data

requirements.txt

@@ -1,4 +1,4 @@
-aiohttp==1.3.5
+aiohttp==2.0.5
 appdirs==1.4.3
 appnope==0.1.0
 async-timeout==1.2.0

setup.py

@@ -30,7 +30,7 @@ def read(*parts):
     long_description=read('README.rst'),
     install_requires=[
         'elasticsearch>=5.0.0,<6.0.0',
-        'aiohttp>=1.3.0,<2.0.0',
+        'aiohttp>=1.3.0',
     ],
     packages=['aioelasticsearch'],
     include_package_data=True,

tox.ini

@@ -1,10 +1,12 @@
 [tox]
 envlist =
-    py3{4,5,6}
+    py3{4,5,6}-aiohttp{1,2}
 skip_missing_interpreters = True
 
 [testenv]
 deps =
+    aiohttp1: aiohttp<2.0.0
+    aiohttp2: aiohttp>=2.0.0
     pytest
     pytest-cov
     flake8