|
1 | 1 | """app.io.py""" |
2 | 2 | import json |
3 | 3 | import pathlib |
4 | | -from typing import Dict, Union |
| 4 | +from typing import Dict, List, Union |
| 5 | + |
| 6 | +import aiofiles |
5 | 7 |
|
6 | 8 | HERE = pathlib.Path(__file__) |
7 | 9 | DATA = HERE.joinpath("..", "data").resolve() |
8 | 10 |
|
9 | 11 |
|
10 | 12 | def save( |
11 | | - name: str, content: Union[str, Dict], write_mode: str = "w", indent: int = 2, **json_dumps_kwargs |
| 13 | + name: str, content: Union[str, Dict, List], write_mode: str = "w", indent: int = 2, **json_dumps_kwargs |
12 | 14 | ) -> pathlib.Path: |
13 | 15 | """Save content to a file. If content is a dictionary, use json.dumps().""" |
14 | 16 | path = DATA / name |
15 | | - if isinstance(content, dict): |
| 17 | + if isinstance(content, (dict, list)): |
16 | 18 | content = json.dumps(content, indent=indent, **json_dumps_kwargs) |
17 | 19 | with open(DATA / name, mode=write_mode) as f_out: |
18 | 20 | f_out.write(content) |
19 | 21 | return path |
20 | 22 |
|
21 | 23 |
|
22 | | -def load(name: str, **json_kwargs) -> Union[str, Dict]: |
| 24 | +def load(name: str, **json_kwargs) -> Union[str, Dict, List]: |
23 | 25 | """Loads content from a file. If file ends with '.json', call json.load() and return a Dictionary.""" |
24 | 26 | path = DATA / name |
25 | 27 | with open(path) as f_in: |
26 | 28 | if path.suffix == ".json": |
27 | 29 | return json.load(f_in, **json_kwargs) |
28 | 30 | return f_in.read() |
| 31 | + |
| 32 | + |
| 33 | +class AIO: |
| 34 | + """Asynsc compatible file io operations.""" |
| 35 | + |
| 36 | + @classmethod |
| 37 | + async def save( |
| 38 | + cls, name: str, content: Union[str, Dict, List], write_mode: str = "w", indent: int = 2, **json_dumps_kwargs |
| 39 | + ): |
| 40 | + """Save content to a file. If content is a dictionary, use json.dumps().""" |
| 41 | + path = DATA / name |
| 42 | + if isinstance(content, (dict, list)): |
| 43 | + content = json.dumps(content, indent=indent, **json_dumps_kwargs) |
| 44 | + async with aiofiles.open(DATA / name, mode=write_mode) as f_out: |
| 45 | + await f_out.write(content) |
| 46 | + return path |
| 47 | + |
| 48 | + @classmethod |
| 49 | + async def load(cls, name: str, **json_kwargs) -> Union[str, Dict, List]: |
| 50 | + """Loads content from a file. If file ends with '.json', call json.load() and return a Dictionary.""" |
| 51 | + path = DATA / name |
| 52 | + async with aiofiles.open(path) as f_in: |
| 53 | + content = await f_in.read() |
| 54 | + if path.suffix == ".json": |
| 55 | + content = json.loads(content, **json_kwargs) |
| 56 | + return content |
0 commit comments