if is_py2:
from urllib.parse import quote, unquote, quote_plus, unquote_plus, urlencode
from urllib.parse import urlparse, urlunparse, urljoin, urlsplit, urldefrag
- from urllib2 import parse_http_list
+ from urllib2 import parse_http_list # pylint: disable=import-error
import http.cookiejar
from http.cookies import Morsel
from io import StringIO
"""Returns true if :attr:`status_code` is 'OK'."""
return self.ok
- def __bool__(self):
- """Returns true if :attr:`status_code` is 'OK'."""
- return self.ok
-
def __iter__(self):
"""Allows you to use a response as an iterator."""
return self.iter_content(128)
self._mTotalChars += 1
# order is valid
if order < self._mTableSize:
- if 512 > self._mCharToFreqOrder[order]:
+ if 512 > self._mCharToFreqOrder[order]: # pylint: disable=unsubscriptable-object
self._mFreqChars += 1
def get_confidence(self):
'mTypicalPositiveRatio': 0.976601,
'keepEnglishLetter': False,
'charsetName': "MacCyrillic"
-};
+}
Ibm866Model = {
'charToOrderMap': IBM866_CharToOrderMap,
self._prepare_conn(conn)
-class HTTPSConnection(HTTPConnection):
+class HTTPSConnection(HTTPConnection): # pylint: disable=function-redefined
default_port = port_by_scheme['https']
def __init__(self, host, port=None, key_file=None, cert_file=None,
return self
def __exit__(self, exc_type, exc_val, exc_tb):
- self.close()
+ self.close() # pylint: disable=too-many-function-args
# Return False to re-raise any potential exceptions
return False
- def close():
+ def close(self):
"""
Close all pooled connections and disable the pool.
"""
except ImportError:
from http.client import HTTPSConnection
from logging import getLogger
-from ntlm import ntlm
+from ntlm import ntlm # pylint: disable=import-error
from urllib3 import HTTPSConnectionPool
'''
try:
- from ndg.httpsclient.ssl_peer_verification import SUBJ_ALT_NAME_SUPPORT
- from ndg.httpsclient.subj_alt_name import SubjectAltName as BaseSubjectAltName
+ from ndg.httpsclient.ssl_peer_verification import SUBJ_ALT_NAME_SUPPORT # pylint: disable=import-error
+ from ndg.httpsclient.subj_alt_name import SubjectAltName as BaseSubjectAltName # pylint: disable=import-error
except SyntaxError as e:
raise ImportError(e)
import OpenSSL.SSL
from pyasn1.codec.der import decoder as der_decoder
from pyasn1.type import univ, constraint
-from socket import _fileobject, timeout
+from socket import _fileobject, timeout # pylint: disable=no-name-in-module
import ssl
import select
for k in self:
yield (k, self[k])
- def update(*args, **kwds):
+ def update(self, *args, **kwds):
'''od.update(E, **F) -> None. Update od from dict/iterable E and F.
If E is a dict instance, does: for k in E: od[k] = E[k]
def urlopen(self, method, url, body=None, headers=None,
encode_multipart=True, multipart_boundary=None,
**kw): # Abstract
- raise NotImplemented("Classes extending RequestMethods must implement "
+ raise NotImplementedError("Classes extending RequestMethods must implement "
"their own ``urlopen`` method.")
def request(self, method, url, fields=None, headers=None, **urlopen_kw):
except BaseSSLError as e:
# FIXME: Is there a better way to differentiate between SSLErrors?
- if 'read operation timed out' not in str(e): # Defensive:
+ if 'read operation timed out' not in str(e): # pylint: disable=not-callable
+ # Defensive:
# This shouldn't happen but just in case we're missing an edge
# case, let's avoid swallowing SSL errors.
raise
"""
if self.total is False and error:
# Disabled, indicate to re-raise the error.
- raise six.reraise(type(error), error, _stacktrace)
+ raise six.reraise(type(error), error, _stacktrace) # pylint: disable=raising-bad-type
total = self.total
if total is not None:
if error and self._is_connection_error(error):
# Connect retry?
if connect is False:
- raise six.reraise(type(error), error, _stacktrace)
+ raise six.reraise(type(error), error, _stacktrace) # pylint: disable=raising-bad-type
elif connect is not None:
connect -= 1
_observed_errors += 1
elif error and self._is_read_error(error):
# Read retry?
if read is False:
- raise six.reraise(type(error), error, _stacktrace)
+ raise six.reraise(type(error), error, _stacktrace) # pylint: disable=raising-bad-type
elif read is not None:
read -= 1
_observed_errors += 1
except ImportError:
import sys
- class SSLContext(object): # Platform-specific: Python 2 & 3.1
+ class SSLContext(object): # pylint: disable=function-redefined
+ # Platform-specific: Python 2 & 3.1
supports_set_ciphers = ((2, 7) <= sys.version_info < (3,) or
(3, 2) <= sys.version_info)
# and returns str instead of unicode somewhere.
from io import StringIO
else:
- from cStringIO import StringIO
+ from cStringIO import StringIO # pylint: disable=import-error
class RangeError(IOError):
"""Error raised when an unsatisfiable range is requested."""
mtype = mimetypes.guess_type(file)[0]
if host:
host, port = urllib.splitport(host)
- if port or socket.gethostbyname(host) not in self.get_names():
+ if port or socket.gethostbyname(host) not in self.get_names(): # pylint: disable=unsupported-membership-test
raise URLError('file not on local host')
fo = open(localfile,'rb')
brange = req.headers.get('Range',None)
path, attrs = splitattr(req.get_selector())
dirs = path.split('/')
dirs = map(unquote, dirs)
- dirs, file = dirs[:-1], dirs[-1]
+ dirs, file = dirs[:-1], dirs[-1] # pylint: disable=unsubscriptable-object
if dirs and not dirs[0]:
dirs = dirs[1:]
try:
if retrlen is not None and retrlen >= 0:
headers += "Content-Length: %d\n" % retrlen
sf = StringIO(headers)
- headers = mimetools.Message(sf)
+ headers = mimetools.Message(sf) # pylint: disable=undefined-variable
return addinfourl(fp, headers, req.get_full_url())
except ftplib.all_errors as msg:
raise IOError('ftp error', msg).with_traceback(sys.exc_info()[2])
# and returns str instead of unicode somewhere.
from io import StringIO
else:
- from cStringIO import StringIO
+ from cStringIO import StringIO # pylint: disable=import-error
from .byterange import range_tuple_normalize, range_tuple_to_header, RangeError
if self.libproxy:
global _libproxy_cache
- if _libproxy_cache is None:
+ if _libproxy_cache is None: # pylint: disable=used-before-assignment
try:
- import libproxy
+ import libproxy # pylint: disable=import-error
_libproxy_cache = libproxy.ProxyFactory()
except:
_libproxy_cache = False
hdrfp = StringIO()
hdrfp.write(self._hdr_dump[statusend:])
hdrfp.seek(0)
- self._parsed_hdr = email.message_from_string(hdrfp)
+ self._parsed_hdr = email.message_from_string(hdrfp) # pylint: disable=undefined-variable
return self._parsed_hdr
hdr = property(_return_hdr_obj)
# mstenner did before here
return (self.fo, self.hdr)
- try:
+ try: # pylint: disable=unreachable
if self.opts.timeout:
old_to = socket.getdefaulttimeout()
socket.setdefaulttimeout(self.opts.timeout)
err.url = self.url
raise err
- except socket.timeout as e:
+ except socket.timeout as e: # pylint: disable=bad-except-order
raise URLGrabError(12, _('Timeout on %s: %s') % (self.url, e))
- err.url = self.url
+ err.url = self.url # pylint: disable=unreachable
raise err
except IOError as e:
raise URLGrabError(4, _('IOError on %s: %s') %(self.url, e))
- err.url = self.url
+ err.url = self.url # pylint: disable=unreachable
raise err
newsize = len(new)
if not self._complete: self._do_grab()
return self.fo.readline()
- i = self._rbuf.find('\n')
+ i = self._rbuf.find('\n') # pylint: disable=unreachable
while i < 0 and not (0 < limit <= len(self._rbuf)):
L = len(self._rbuf)
self._fill_buffer(L + self._rbufsize)
if s[j] in ')]':
if s[j] == ')':
l = tuple(l)
- stk[0].append(l)
- l, stk = stk
+ stk[0].append(l) # pylint: disable=unsubscriptable-object
+ l, stk = stk # pylint: disable=unpacking-non-sequence
i = j = j + 1
elif s[j] in '[(':
stk = l, stk
# We use a version check because python2 also has _thread
import _thread as thread
else:
- import thread
+ import thread # pylint: disable=import-error
try:
import urllib.parse as urlparse
self.increment_mirror(gr, action)
if action and action.get('fail', 0):
sys.exc_info()[1].errors = gr.errors
- raise
+ raise # pylint: disable=misplaced-bare-raise
def increment_mirror(self, gr, action={}):
"""Tell the mirror object increment the mirror index
# We use a version check because python2 also has _thread
import _thread as thread
else:
- import thread
+ import thread # pylint: disable=import-error
# Code from http://mail.python.org/pipermail/python-list/2000-May/033365.html
def terminal_width(fd=1):