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

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

2 

3import json 

4import os 

5from pathlib import Path 

6from typing import TYPE_CHECKING, Any, Optional 

7 

8try: 

9 import dotenv 

10except ImportError: 

11 pass 

12else: 

13 _ = dotenv.load_dotenv() 

14 

15if TYPE_CHECKING: 

16 import httpx 

17 

18 

19def get_report_path( 

20 item_id: str, 

21 version: str, 

22) -> Path: 

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

24 

25 

26def get_tool_report_path( 

27 item_id: str, 

28 version: str, 

29 tool_name: str, 

30 tool_version: str, 

31): 

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

33 if "_" in tool_name: 

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

35 

36 if "_" in tool_version: 

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

38 

39 return ( 

40 get_report_path(item_id, version) 

41 / "reports" 

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

43 ) 

44 

45 

46def get_all_tool_report_paths( 

47 item_id: str, 

48 version: str, 

49): 

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

51 

52 

53def get_summary_data(item_id: str, version: str) -> Optional[dict[str, Any]]: 

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

55 summary_file_path = get_summary_file_path(item_id, version) 

56 if not summary_file_path.exists(): 

57 return None 

58 

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

60 return json.load(f) 

61 

62 

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

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

65 

66 

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

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

69 

70 

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

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

73 import httpx 

74 

75 local_path = Path("cache") / sha256 

76 if not local_path.exists(): 

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

78 response = httpx.get( 

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

80 ).raise_for_status() 

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

82 _ = f.write(response.content) 

83 

84 return local_path 

85 

86 

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

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

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

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

91 

92 

93def raise_for_status_discretely(response: "httpx.Response"): 

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

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

96 """ 

97 import httpx 

98 

99 http_error_msg = "" 

100 reason = response.reason_phrase 

101 

102 discrete_url = response.url.copy_with( 

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

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

105 ) 

106 

107 if 400 <= response.status_code < 500: 

108 http_error_msg = ( 

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

110 ) 

111 

112 elif 500 <= response.status_code < 600: 

113 http_error_msg = ( 

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

115 ) 

116 

117 if http_error_msg: 

118 raise httpx.HTTPError(http_error_msg)