Skip to content
Merged
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
add async save/read
  • Loading branch information
Kilo59 committed Apr 29, 2020
commit da174120dcce189e31e3546413714e3e87b968a2
36 changes: 32 additions & 4 deletions app/io.py
Original file line number Diff line number Diff line change
@@ -1,28 +1,56 @@
"""app.io.py"""
import json
import pathlib
from typing import Dict, Union
from typing import Dict, List, Union

import aiofiles

HERE = pathlib.Path(__file__)
DATA = HERE.joinpath("..", "data").resolve()


def save(
name: str, content: Union[str, Dict], write_mode: str = "w", indent: int = 2, **json_dumps_kwargs
name: str, content: Union[str, Dict, List], write_mode: str = "w", indent: int = 2, **json_dumps_kwargs
) -> pathlib.Path:
"""Save content to a file. If content is a dictionary, use json.dumps()."""
path = DATA / name
if isinstance(content, dict):
if isinstance(content, (dict, list)):
content = json.dumps(content, indent=indent, **json_dumps_kwargs)
with open(DATA / name, mode=write_mode) as f_out:
f_out.write(content)
return path


def load(name: str, **json_kwargs) -> Union[str, Dict]:
def load(name: str, **json_kwargs) -> Union[str, Dict, List]:
"""Loads content from a file. If file ends with '.json', call json.load() and return a Dictionary."""
path = DATA / name
with open(path) as f_in:
if path.suffix == ".json":
return json.load(f_in, **json_kwargs)
return f_in.read()


class AIO:
"""Asynsc compatible file io operations."""

@classmethod
async def save(
cls, name: str, content: Union[str, Dict, List], write_mode: str = "w", indent: int = 2, **json_dumps_kwargs
):
"""Save content to a file. If content is a dictionary, use json.dumps()."""
path = DATA / name
if isinstance(content, (dict, list)):
content = json.dumps(content, indent=indent, **json_dumps_kwargs)
async with aiofiles.open(DATA / name, mode=write_mode) as f_out:
await f_out.write(content)
return path

@classmethod
async def load(cls, name: str, **json_kwargs) -> Union[str, Dict, List]:
"""Loads content from a file. If file ends with '.json', call json.load() and return a Dictionary."""
path = DATA / name
async with aiofiles.open(path) as f_in:
content = await f_in.read()
if path.suffix == ".json":
content = json.loads(content, **json_kwargs)
return content