# Ideally we'd run the lint stuff on the latest Python
# version supported. But Python 3.7 requires slower-to-start VMs,
# so we run it on 3.6 to minimize total CI run time.
- - if [[ $TRAVIS_PYTHON_VERSION == '3.6' ]]; then travis_retry pip install flake8 mypy==0.630 black; fi
+ - if [[ $TRAVIS_PYTHON_VERSION == '3.6' ]]; then travis_retry pip install flake8 mypy==0.701 black; fi
# We run docs on py37 because we need some bug fixes for introspection of annotations.
- if [[ $TRAVIS_PYTHON_VERSION == '3.7' ]]; then travis_retry pip install -r docs/requirements.txt; fi
# On travis the extension should always be built
import collections
import functools
import logging
-import pycurl # type: ignore
+import pycurl
import threading
import time
from io import BytesIO
self, max_clients: int = 10, defaults: Dict[str, Any] = None
) -> None:
super(CurlAsyncHTTPClient, self).initialize(defaults=defaults)
- self._multi = pycurl.CurlMulti()
+ # Typeshed is incomplete for CurlMulti, so just use Any for now.
+ self._multi = pycurl.CurlMulti() # type: Any
self._multi.setopt(pycurl.M_TIMERFUNCTION, self._set_timeout)
self._multi.setopt(pycurl.M_SOCKETFUNCTION, self._handle_socket)
self._curls = [self._curl_create() for i in range(max_clients)]
started += 1
curl = self._free_list.pop()
(request, callback, queue_start_time) = self._requests.popleft()
- curl.info = {
+ # TODO: Don't smuggle extra data on an attribute of the Curl object.
+ curl.info = { # type: ignore
"headers": httputil.HTTPHeaders(),
"buffer": BytesIO(),
"request": request,
}
try:
self._curl_setup_request(
- curl, request, curl.info["buffer"], curl.info["headers"]
+ curl,
+ request,
+ curl.info["buffer"], # type: ignore
+ curl.info["headers"], # type: ignore
)
except Exception as e:
# If there was an error in setup, pass it on
def _finish(
self, curl: pycurl.Curl, curl_error: int = None, curl_message: str = None
) -> None:
- info = curl.info
- curl.info = None
+ info = curl.info # type: ignore
+ curl.info = None # type: ignore
self._multi.remove_handle(curl)
self._free_list.append(curl)
buffer = info["buffer"]
request_buffer = BytesIO(utf8(request.body or ""))
def ioctl(cmd: int) -> None:
- if cmd == curl.IOCMD_RESTARTREAD:
+ if cmd == curl.IOCMD_RESTARTREAD: # type: ignore
request_buffer.seek(0)
curl.setopt(pycurl.READFUNCTION, request_buffer.read)
self.__failure = None
raise_exc_info(failure)
- def run(self, result: unittest.TestResult = None) -> unittest.TestCase:
+ def run(self, result: unittest.TestResult = None) -> Optional[unittest.TestResult]:
ret = super(AsyncTestCase, self).run(result)
# As a last resort, if an exception escaped super.run() and wasn't
# re-raised in tearDown, raise it here. This will cause the