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 Any 

26 

27import dotenv 

28import requests 

29import requests.exceptions 

30import yaml 

31 

32logger = logging.getLogger(__name__) 

33 

34 

35class RemoteFileLoader: 

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

37 

38 Args: 

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

40 

41 Attributes: 

42 available_formats: File formats with ad-hoc parser available. 

43 parser: Parser selected for this object. 

44 

45 """ 

46 

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

48 parser: str | None = None 

49 

50 def __init__(self, parser: str | None = None) -> None: 

51 if parser in self.available_formats: 

52 self.parser = parser 

53 

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

55 if self.parser == "yaml": 

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

57 if self.parser == "ini": 

58 config = configparser.ConfigParser() 

59 try: 

60 config.read_string(content) 

61 except configparser.MissingSectionHeaderError: 

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

63 config.read_string(content) 

64 return config 

65 if self.parser == "env": 

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

67 if self.parser == "json": 

68 return json.loads(content) 

69 # Only return content, no parsing 

70 return content 

71 

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

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

74 

75 Args: 

76 url: URL of the remote file to fetch. 

77 

78 Raises: 

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

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

81 

82 """ 

83 try: 

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

85 if r.status_code == 200: 

86 return self.__parse(r.text) 

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

88 except requests.exceptions.HTTPError as ex: 

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

90 raise ex 

91 except requests.exceptions.Timeout as ex: 

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

93 raise ex