Coverage for src/extratools_core/jsontools.py: 57%

1from csv import DictWriter 

2from io import StringIO 

3from pathlib import Path 

4from typing import Any 

5 

6 

7def flatten(data: Any) -> Any: 

8 def flatten_rec(data: Any, path: str) -> None: 

9 if isinstance(data, dict): 

10 for k, v in data.items(): 

11 flatten_rec(v, path + (f".{k}" if path else k)) 

12 elif isinstance(data, list): 

13 for i, v in enumerate(data): 

14 flatten_rec(v, path + f"[{i}]") 

15 else: 

16 flatten_dict[path or "."] = data 

17 

18 flatten_dict: dict[str, Any] = {} 

19 flatten_rec(data, "") 

20 return flatten_dict 

21 

22 

23def json_to_csv( 

24 data: dict[str, dict[str, Any]] | list[dict[str, Any]], 

25 /, 

26 csv_path: Path | str | None = None, 

27 *, 

28 key_field_name: str = "_key", 

29) -> str: 

30 if isinstance(data, dict): 

31 data = [ 

32 { 

33 # In case there is already a key field in each record, 

34 # the new key field will be overwritten. 

35 # It is okay though as the existing key field is likely 

36 # serving the purpose of containing keys. 

37 key_field_name: key, 

38 **value, 

39 } 

40 for key, value in data.items() 

41 ] 

42 

43 fields: set[str] = set() 

44 for record in data: 

45 fields.update(record.keys()) 

46 

47 sio = StringIO() 

48 

49 writer = DictWriter(sio, fieldnames=fields) 

50 writer.writeheader() 

51 writer.writerows(data) 

52 

53 csv_str: str = sio.getvalue() 

54 

55 if csv_path: 

56 Path(csv_path).write_text(csv_str) 

57 

58 return csv_str