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/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 .compat import AIOHTTP_2  # isort:skip
+
+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,
@@ -49,15 +58,23 @@ def __init__(
 
         self.session = kwargs.get('session')
         if self.session is None:
+            connector_kwargs = {
+                'limit': maxsize,
+                'use_dns_cache': kwargs.get('use_dns_cache', False),
+                'ssl_context': ssl_context,
+                'verify_ssl': self.verify_certs,
+                'loop': self.loop,
+            }
+            session_kwargs = {'auth': self.http_auth}
+
+            if AIOHTTP_2:
+                session_kwargs['conn_timeout'] = None
+            else:
+                connector_kwargs['conn_timeout'] = None
+
             self.session = aiohttp.ClientSession(
-                auth=self.http_auth,
-                connector=aiohttp.TCPConnector(
-                    limit=maxsize,
-                    use_dns_cache=kwargs.get('use_dns_cache', False),
-                    verify_ssl=self.verify_certs,
-                    conn_timeout=None,
-                    loop=self.loop,
-                ),
+                connector=aiohttp.TCPConnector(**connector_kwargs),
+                **session_kwargs
             )
 
     def close(self):
@@ -78,17 +95,22 @@ def perform_request(self, method, url, params=None, body=None, timeout=None, ign
 
             duration = self.loop.time() - start
 
-        except asyncio.TimeoutError as exc:
+        except ssl.CertificateError as exc:
             self.log_request_fail(method, url, url_path, body, self.loop.time() - start, exception=exc)  # noqa
-            raise ConnectionTimeout('TIMEOUT', str(exc), exc)
+            raise SSLError('N/A', str(exc), exc)
 
-        except FingerprintMismatch as exc:
+        except asyncio.TimeoutError 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)
+            raise ConnectionTimeout('TIMEOUT', 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)
+            _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:

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