Coverage for src/ensembl/utils/rloader.py: 100%

1# See the NOTICE file distributed with this work for additional information 

2# regarding copyright ownership. 

3# 

4# Licensed under the Apache License, Version 2.0 (the "License"); 

5# you may not use this file except in compliance with the License. 

6# You may obtain a copy of the License at 

7# 

8# http://www.apache.org/licenses/LICENSE-2.0 

9# 

10# Unless required by applicable law or agreed to in writing, software 

11# distributed under the License is distributed on an "AS IS" BASIS, 

12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

13# See the License for the specific language governing permissions and 

14# limitations under the License. 

15"""Allow to seamlessly load / read the content of a remote file as if it was located locally.""" 

16 

17from __future__ import annotations 

18 

19__all__ = ["RemoteFileLoader"] 

20 

21import configparser 

22import json 

23import logging 

24from io import StringIO 

25from typing import Optional, Any 

26 

27import dotenv 

28import requests 

29import requests.exceptions 

30import yaml 

31 

32 

33logger = logging.getLogger(__name__) 

34 

35 

36class RemoteFileLoader: 

37 """Loads remote files, allowing specific format parsing options. 

38 

39 Args: 

40 parser: Parser to use for this object. Default: `None` (no format-specific parsing done). 

41 

42 Attributes: 

43 available_formats: File formats with ad-hoc parsers available. 

44 parser: Parser selected for this object. 

45 

46 """ 

47 

48 available_formats: set[str] = {"yaml", "ini", "env", "json"} 

49 parser: Optional[str] = None 

50 

51 def __init__(self, parser: Optional[str] = None) -> None: 

52 if parser in self.available_formats: 

53 self.parser = parser 

54 

55 def __parse(self, content: str) -> Any: 

56 if self.parser == "yaml": 

57 return yaml.load(content, yaml.SafeLoader) 

58 if self.parser == "ini": 

59 config = configparser.ConfigParser() 

60 try: 

61 config.read_string(content) 

62 except configparser.MissingSectionHeaderError: 

63 content = "[DEFAULT]\n" + content 

64 config.read_string(content) 

65 return config 

66 if self.parser == "env": 

67 return dotenv.dotenv_values(stream=StringIO(content)) 

68 if self.parser == "json": 

69 return json.loads(content) 

70 # Only return content, no parsing 

71 return content 

72 

73 def r_open(self, url: str) -> Any: 

74 """Returns the parsed remote file from the given URL. 

75 

76 Args: 

77 url: URL of the remote file to fetch. 

78 

79 Raises: 

80 requests.exception.HTTPError: If loading or requesting the given URL returned an error. 

81 requests.exception.Timeout: If a timeout was raised whilst requesting the given URL. 

82 

83 """ 

84 try: 

85 r = requests.get(url, timeout=120) 

86 if r.status_code == 200: 

87 return self.__parse(r.text) 

88 raise requests.exceptions.HTTPError(response=r) 

89 except requests.exceptions.HTTPError as ex: 

90 logger.exception(f"Error with request to {url}: {ex}") 

91 raise ex 

92 except requests.exceptions.Timeout as ex: 

93 logger.exception(f"Request timed out {url}: {ex}") 

94 raise ex