Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/werkzeug/_internal.py: 48%

1from __future__ import annotations 

2 

3import datetime as dt 

4import logging 

5import re 

6import typing as t 

7import wsgiref.types as wt 

8 

9if t.TYPE_CHECKING: 

10 from .wrappers.request import Request 

11 

12T = t.TypeVar("T") 

13_logger: logging.Logger | None = None 

14 

15 

16class _Missing: 

17 def __repr__(self) -> str: 

18 return "no value" 

19 

20 def __reduce__(self) -> str: 

21 return "_missing" 

22 

23 

24_missing = _Missing() 

25 

26 

27def _wsgi_decoding_dance(s: str) -> str: 

28 return s.encode("latin1").decode(errors="replace") 

29 

30 

31def _wsgi_encoding_dance(s: str) -> str: 

32 return s.encode().decode("latin1") 

33 

34 

35def _get_environ(obj: wt.WSGIEnvironment | Request) -> wt.WSGIEnvironment: 

36 env = getattr(obj, "environ", obj) 

37 assert isinstance(env, dict), ( 

38 f"{type(obj).__name__!r} is not a WSGI environment (has to be a dict)" 

39 ) 

40 return env 

41 

42 

43def _has_level_handler(logger: logging.Logger) -> bool: 

44 """Check if there is a handler in the logging chain that will handle 

45 the given logger's effective level. 

46 """ 

47 level = logger.getEffectiveLevel() 

48 current: logging.Logger | None = logger 

49 

50 while current: 

51 if any(handler.level <= level for handler in current.handlers): 

52 return True 

53 

54 if not current.propagate: 

55 break 

56 

57 current = current.parent 

58 

59 return False 

60 

61 

62def _log(type: str, message: str, *args: t.Any, **kwargs: t.Any) -> None: 

63 """Log a message to the 'werkzeug' logger. 

64 

65 The logger is created the first time it is needed. If there is no 

66 level set, it is set to :data:`logging.INFO`. If there is no handler 

67 for the logger's effective level, a :class:`logging.StreamHandler` 

68 is added. 

69 """ 

70 global _logger 

71 

72 if _logger is None: 

73 _logger = logging.getLogger("werkzeug") 

74 

75 if _logger.level == logging.NOTSET: 

76 _logger.setLevel(logging.INFO) 

77 

78 if not _has_level_handler(_logger): 

79 _logger.addHandler(logging.StreamHandler()) 

80 

81 getattr(_logger, type)(message.rstrip(), *args, **kwargs) 

82 

83 

84@t.overload 

85def _dt_as_utc(value: None) -> None: ... 

86@t.overload 

87def _dt_as_utc(value: dt.datetime) -> dt.datetime: ... 

88def _dt_as_utc(value: dt.datetime | None) -> dt.datetime | None: 

89 if value is None: 

90 return None 

91 

92 if value.tzinfo is None: 

93 return value.replace(tzinfo=dt.UTC) 

94 elif value.tzinfo != dt.UTC: 

95 return value.astimezone(dt.UTC) 

96 

97 return value 

98 

99 

100_plain_int_re = { 

101 10: re.compile(r"-?[0-9]+", re.ASCII), 

102 16: re.compile(r"-?[0-9a-f]+", re.ASCII | re.IGNORECASE), 

103} 

104 

105 

106def _plain_int(value: str, base: t.Literal[10, 16] = 10) -> int: 

107 """Parse an int only if it valid ASCII characters for the base. 

108 

109 This disallows ``+``, ``_``, ``0x``, and non-ASCII digits, which are 

110 accepted by ``int`` but are not allowed in HTTP header values. 

111 

112 Any surrounding spaces and tabs are stripped. 

113 """ 

114 value = value.strip(" \t") 

115 

116 if _plain_int_re[base].fullmatch(value) is None: 

117 raise ValueError(value) 

118 

119 return int(value, base)