Coverage for src/backoffice/utils_pure.py: 47%

1"""utility functions available in backoffice without dependencies""" 

2from __future__ import annotations 

3 

4import json 

5import os 

6from pathlib import Path 

7from typing import TYPE_CHECKING, Any 

8 

9try: 

10 import dotenv 

11except ImportError: 

12 pass 

13else: 

14 _ = dotenv.load_dotenv() 

15 

16if TYPE_CHECKING: 

17 import httpx 

18 

19 

20def get_report_path( 

21 item_id: str, 

22 version: str, 

23) -> Path: 

24 return Path(os.getenv("REPORTS", "reports")) / item_id.replace(":", "_") / version 

25 

26 

27def get_tool_report_path( 

28 item_id: str, 

29 version: str, 

30 tool_name: str, 

31 tool_version: str, 

32): 

33 """Get the path to the report for a specific item version and tool.""" 

34 if "_" in tool_name: 

35 raise ValueError("Underscore not allowed in tool_name") 

36 

37 if "_" in tool_version: 

38 raise ValueError("Underscore not allowed in tool_version") 

39 

40 return ( 

41 get_report_path(item_id, version) 

42 / "reports" 

43 / f"{tool_name}_{tool_version}.json" 

44 ) 

45 

46 

47def get_all_tool_report_paths( 

48 item_id: str, 

49 version: str, 

50): 

51 return list((get_report_path(item_id, version) / "reports").glob("*.json")) 

52 

53 

54def get_summary_data(item_id: str, version: str) -> dict[str, Any] | None: 

55 """Get the summary data of a specific item version.""" 

56 summary_file_path = get_summary_file_path(item_id, version) 

57 if not summary_file_path.exists(): 

58 return None 

59 

60 with summary_file_path.open(encoding="utf-8") as f: 

61 return json.load(f) 

62 

63 

64def get_summary_file_path(item_id: str, version: str) -> Path: 

65 return get_report_path(item_id, version) / "summary.json" 

66 

67 

68def get_log_file(item_id: str, version: str) -> Path: 

69 return get_report_path(item_id, version) / "log.txt" 

70 

71 

72def cached_download(url: str, sha256: str) -> Path: 

73 """Download a file from the given URL and cache it locally.""" 

74 import httpx 

75 

76 local_path = Path("cache") / sha256 

77 if not local_path.exists(): 

78 local_path.parent.mkdir(parents=True, exist_ok=True) 

79 response = httpx.get( 

80 url, timeout=float(os.environ.get("HTTP_TIMEOUT", "30")) 

81 ).raise_for_status() 

82 with local_path.open("wb") as f: 

83 _ = f.write(response.content) 

84 

85 return local_path 

86 

87 

88def get_rdf_content_from_id(item_id: str, version: str) -> dict[str, Any]: 

89 """Get the RDF file content of a specific item version.""" 

90 with get_summary_file_path(item_id, version).open() as f: 

91 return json.load(f)["rdf_content"] 

92 

93 

94def raise_for_status_discretely(response: httpx.Response): 

95 """Raises :class:`httpx.HTTPError` for 4xx or 5xx responses, 

96 **but** hides any query and userinfo from url to avoid leaking sensitive data. 

97 """ 

98 import httpx 

99 

100 http_error_msg = "" 

101 reason = response.reason_phrase 

102 

103 discrete_url = response.url.copy_with( 

104 query=(b"***query*hidden***" if response.url.query else b""), 

105 userinfo=(b"***userinfo*hidden***" if response.url.userinfo else b""), 

106 ) 

107 

108 if 400 <= response.status_code < 500: 

109 http_error_msg = ( 

110 f"{response.status_code} Client Error: {reason} for url: {discrete_url}" 

111 ) 

112 

113 elif 500 <= response.status_code < 600: 

114 http_error_msg = ( 

115 f"{response.status_code} Server Error: {reason} for url: {discrete_url}" 

116 ) 

117 

118 if http_error_msg: 

119 raise httpx.HTTPError(http_error_msg)