1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
|
"""Functions for disk IO."""
from __future__ import annotations
import io
import json
import os
import typing as t
from .encoding import (
ENCODING,
to_bytes,
to_text,
)
def read_json_file(path: str) -> t.Any:
"""Parse and return the json content from the specified path."""
return json.loads(read_text_file(path))
def read_text_file(path: str) -> str:
"""Return the contents of the specified path as text."""
return to_text(read_binary_file(path))
def read_binary_file(path: str) -> bytes:
"""Return the contents of the specified path as bytes."""
with open_binary_file(path) as file_obj:
return file_obj.read()
def make_dirs(path: str) -> None:
"""Create a directory at path, including any necessary parent directories."""
os.makedirs(to_bytes(path), exist_ok=True)
def write_json_file(path: str,
content: t.Any,
create_directories: bool = False,
formatted: bool = True,
encoder: t.Optional[t.Type[json.JSONEncoder]] = None,
) -> str:
"""Write the given json content to the specified path, optionally creating missing directories."""
text_content = json.dumps(content,
sort_keys=formatted,
indent=4 if formatted else None,
separators=(', ', ': ') if formatted else (',', ':'),
cls=encoder,
) + '\n'
write_text_file(path, text_content, create_directories=create_directories)
return text_content
def write_text_file(path: str, content: str, create_directories: bool = False) -> None:
"""Write the given text content to the specified path, optionally creating missing directories."""
if create_directories:
make_dirs(os.path.dirname(path))
with open_binary_file(path, 'wb') as file_obj:
file_obj.write(to_bytes(content))
def open_text_file(path: str, mode: str = 'r') -> t.IO[str]:
"""Open the given path for text access."""
if 'b' in mode:
raise Exception('mode cannot include "b" for text files: %s' % mode)
return io.open(to_bytes(path), mode, encoding=ENCODING) # pylint: disable=consider-using-with
def open_binary_file(path: str, mode: str = 'rb') -> t.IO[bytes]:
"""Open the given path for binary access."""
if 'b' not in mode:
raise Exception('mode must include "b" for binary files: %s' % mode)
return io.open(to_bytes(path), mode) # pylint: disable=consider-using-with
class SortedSetEncoder(json.JSONEncoder):
"""Encode sets as sorted lists."""
def default(self, o: t.Any) -> t.Any:
"""Return a serialized version of the `o` object."""
if isinstance(o, set):
return sorted(o)
return json.JSONEncoder.default(self, o)
|
