Coverage for src/bioimageio/spec/_internal/url.py: 95%

1from __future__ import annotations 

2 

3from contextlib import nullcontext 

4from typing import Any, ClassVar 

5 

6import httpx 

7import pydantic 

8from loguru import logger 

9from pydantic import RootModel 

10from typing_extensions import Literal, assert_never 

11 

12from . import warning_levels 

13from ._settings import settings 

14from .field_warning import issue_warning 

15from .root_url import RootHttpUrl 

16from .validation_context import get_validation_context 

17 

18 

19def _validate_url(url: str | pydantic.HttpUrl) -> pydantic.HttpUrl: 

20 return _validate_url_impl(url, request_mode="head", timeout=settings.http_timeout) 

21 

22 

23def _validate_url_impl( 

24 url: str | pydantic.HttpUrl, 

25 request_mode: Literal["head", "get_stream", "get"], 

26 timeout: float, 

27) -> pydantic.HttpUrl: 

28 url = str(url) 

29 context = get_validation_context() 

30 if url in context.known_files: 

31 return pydantic.HttpUrl(url) 

32 

33 val_url = url 

34 

35 if url.startswith(("http://example.com", "https://example.com")): 

36 return pydantic.HttpUrl(url) 

37 

38 if url.startswith("https://colab.research.google.com/github/"): 

39 # get requests for colab returns 200 even if the source notebook does not exists. 

40 # We therefore validate the url to the notebbok instead (for github notebooks) 

41 val_url = url.replace( 

42 "https://colab.research.google.com/github/", "https://github.com/" 

43 ) 

44 elif url.startswith("https://colab.research.google.com/"): 

45 # TODO: improve validation of non-github colab urls 

46 issue_warning( 

47 "colab urls currently pass even if the notebook url was not found. Cannot fully validate {value}", 

48 value=url, 

49 severity=warning_levels.INFO, 

50 ) 

51 

52 try: 

53 if request_mode in ("head", "get"): 

54 request_ctxt = nullcontext( 

55 httpx.request( 

56 request_mode.upper(), 

57 val_url, 

58 timeout=timeout, 

59 follow_redirects=True, 

60 ) 

61 ) 

62 elif request_mode == "get_stream": 

63 request_ctxt = httpx.stream( 

64 "GET", val_url, timeout=timeout, follow_redirects=True 

65 ) 

66 else: 

67 assert_never(request_mode) 

68 

69 with request_ctxt as r: 

70 status_code = r.status_code 

71 reason = r.reason_phrase 

72 location = r.headers.get("location") 

73 

74 except ( 

75 httpx.InvalidURL, 

76 httpx.TooManyRedirects, 

77 ) as e: 

78 raise ValueError(f"Invalid URL '{url}': {e}") 

79 except httpx.RequestError as e: 

80 issue_warning( 

81 "Failed to validate URL '{value}': {error}\nrequest: {request}", 

82 value=url, 

83 msg_context={"error": str(e), "request": e.request}, 

84 ) 

85 except Exception as e: 

86 issue_warning( 

87 "Failed to validate URL '{value}': {error}", 

88 value=url, 

89 msg_context={"error": str(e)}, 

90 ) 

91 else: 

92 if status_code == 200 or status_code in (302, 303): # ok 

93 pass 

94 elif status_code in (301, 308): 

95 issue_warning( 

96 "URL redirected ({status_code}): consider updating {value} with new" 

97 + " location: {location}", 

98 value=url, 

99 severity=warning_levels.INFO, 

100 msg_context={ 

101 "status_code": status_code, 

102 "location": location, 

103 }, 

104 ) 

105 elif request_mode == "head": 

106 return _validate_url_impl(url, request_mode="get_stream", timeout=timeout) 

107 elif request_mode == "get_stream": 

108 return _validate_url_impl(url, request_mode="get", timeout=timeout) 

109 elif request_mode == "get": 

110 issue_warning( 

111 "{status_code}: {reason} ({value})", 

112 value=url, 

113 severity=( 

114 warning_levels.INFO 

115 if status_code == 405 # may be returned due to a captcha 

116 else warning_levels.WARNING 

117 ), 

118 msg_context={ 

119 "status_code": status_code, 

120 "reason": reason, 

121 }, 

122 ) 

123 else: 

124 assert_never(request_mode) 

125 

126 context.known_files[url] = None 

127 return pydantic.HttpUrl(url) 

128 

129 

130class HttpUrl(RootHttpUrl): 

131 """A URL with the HTTP or HTTPS scheme.""" 

132 

133 root_model: ClassVar[type[RootModel[Any]]] = RootModel[pydantic.HttpUrl] 

134 _exists: bool | None = None 

135 

136 def _after_validator(self): 

137 value = super()._after_validator() 

138 if get_validation_context().perform_io_checks: 

139 _ = value.exists() 

140 

141 return value 

142 

143 def exists(self): 

144 """True if URL is available""" 

145 if self._exists is None: 

146 ctxt = get_validation_context() 

147 try: 

148 with ctxt.replace(warning_level=warning_levels.WARNING): 

149 self._validated = _validate_url(self._validated) 

150 except Exception as e: 

151 if ctxt.log_warnings: 

152 logger.info(e) 

153 

154 self._exists = False 

155 else: 

156 self._exists = True 

157 

158 return self._exists