Coverage for compiler_admin / services / files.py: 91%

1import json 

2import os 

3from io import TextIOBase 

4from pathlib import Path 

5from typing import TextIO 

6 

7import pandas as pd 

8 

9 

10def read_csv(file_path, **kwargs) -> pd.DataFrame: 

11 """Read a file path or buffer of CSV data into a pandas.DataFrame.""" 

12 return pd.read_csv(file_path, **kwargs) 

13 

14 

15def read_json(file_path): 

16 """Read a file path of JSON data into a python object.""" 

17 

18 if isinstance(file_path, (str, Path)): 

19 with open(file_path, "r") as f: 

20 return json.load(f) 

21 elif isinstance(file_path, (TextIOBase, TextIO)): 21 ↛ 24line 21 didn't jump to line 24 because the condition on line 21 was always true

22 return json.load(file_path) 

23 else: 

24 raise NotImplementedError(f"Input type for file_path not allowed: {type(file_path)}") 

25 

26 

27def write_csv(file_path, data: pd.DataFrame, columns: list[str] = None): 

28 """Write a pandas.DataFrame as CSV to the given path or buffer, with an optional list of columns to write.""" 

29 data.to_csv(file_path, columns=columns, index=False) 

30 

31 

32def write_json(file_path, data): 

33 """Write a python object as JSON to the given path.""" 

34 if isinstance(file_path, (str, Path)): 

35 with open(file_path, "w") as f: 

36 json.dump(data, f, indent=2) 

37 elif isinstance(file_path, (TextIOBase, TextIO)): 37 ↛ 40line 37 didn't jump to line 40 because the condition on line 37 was always true

38 json.dump(data, file_path, indent=2) 

39 else: 

40 raise NotImplementedError(f"Input type for file_path not allowed: {type(file_path)}") 

41 

42 

43class JsonFileCache: 

44 """Very basic in-memory cache of a JSON file.""" 

45 

46 def __init__(self, env_file_path=None): 

47 self._cache = {} 

48 self._path = None 

49 

50 if env_file_path: 50 ↛ 53line 50 didn't jump to line 53 because the condition on line 50 was always true

51 p = os.environ.get(env_file_path) 

52 self._path = Path(p) if p else None 

53 if self._path and self._path.exists(): 

54 self._cache.update(read_json(self._path)) 

55 

56 def __contains__(self, key): 

57 return key in self._cache 

58 

59 def __getitem__(self, key): 

60 return self._cache.get(key) 

61 

62 def __setitem__(self, key, value): 

63 self._cache[key] = value 

64 

65 def get(self, key, default=None): 

66 return self._cache.get(key, default)