Passed
Push — develop ( 0bedf3...7270f6 )
by Dean
05:49 queued 02:46
created

RequestsFilter.filter()   A

Complexity

Conditions 3

Size

Total Lines 18

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 1
CRAP Score 9.762
Metric Value
cc 3
dl 0
loc 18
ccs 1
cts 11
cp 0.0909
crap 9.762
rs 9.4285
1 1
from logging import Filter
2 1
from requests import RequestException
3 1
import logging
4
5
6 1
class RequestsLogFilter(Filter):
7 1
    def filter(self, record):
8
        if self.is_dropped_connection(record):
9
            # Change record level to debug
10
            record.levelno = logging.DEBUG
11
            record.levelname = 'DEBUG'
12
13
            # Retrieve logger for record
14
            logger = logging.getLogger(record.name)
15
16
            # Check if the logger has debug logging enabled
17
            return logger.isEnabledFor(logging.DEBUG)
18
19
        return True
20
21 1
    @staticmethod
22
    def is_dropped_connection(record):
23
        if record.levelno != logging.INFO:
24
            return False
25
26
        if record.name != 'requests.packages.urllib3.connectionpool':
27
            return False
28
29
        if record.msg and not record.msg.startswith('Resetting dropped connection:'):
30
            return False
31
32
        return True
33
34
35 1
class RequestsReportFilter(Filter):
36 1
    def filter(self, record):
37
        if self.is_requests_message(record):
38
            return False
39
40
        return True
41
42 1
    @staticmethod
43
    def is_requests_message(record):
44
        if record.levelno < logging.WARNING:
45
            return False
46
47
        if not record.exc_info or len(record.exc_info) != 3:
48
            return False
49
50
        exc_type, _, _ = record.exc_info
51
52
        if not exc_type or not issubclass(exc_type, RequestException):
53
            return False
54
55
        return True
56