from asyncio import Lock
import inspect
import json
from pathlib import Path
from typing import Optional
import yaml
from vyra_base.helper._aiopath import AsyncPath
from vyra_base.helper.file_lock import (
get_lock_for_file,
release_lock_for_file,
get_lock_for_file_sync,
release_lock_for_file_sync
)
[Doku]
class FileWriter:
"""File writer.
Writes Python objects to a file with both async and sync methods.
"""
@staticmethod
async def _resolve_async_open(open_result):
"""Normalize AsyncPath.open return type across runtimes and mocks."""
if inspect.isawaitable(open_result):
return await open_result
return open_result
# ========================================================================
# ASYNC METHODS
# ========================================================================
[Doku]
@classmethod
async def write_json_file(cls, file: Path, file_content: dict) -> Optional[bool]:
"""
Writes a JSON file from a dictionary (asynchronous).
:param file: The path to the file to write.
:type file: Path
:param file_content: The dictionary to serialize as JSON.
:type file_content: dict
:returns: True if writing finished successfully, else False.
:rtype: bool
"""
try:
lock: Lock = await get_lock_for_file(file)
async with lock:
writer_cm = await cls._resolve_async_open(AsyncPath(file).open(mode='w'))
async with writer_cm as writer:
await writer.write(json.dumps(file_content, indent=4))
return True
except (IOError, UnicodeDecodeError, json.decoder.JSONDecodeError,
TypeError, FileNotFoundError) as error:
if isinstance(error, IOError):
raise IOError('An unexpected io error occured!')
if isinstance(error, UnicodeDecodeError):
raise UnicodeDecodeError('A decoding error occured!', b'', 0, 0, '')
if isinstance(error, json.decoder.JSONDecodeError):
raise json.decoder.JSONDecodeError('JSON decoding error:', '', 0)
if isinstance(error, TypeError):
raise TypeError("An unexpected type error uccured!")
if isinstance(error, FileNotFoundError):
raise FileNotFoundError("File not found error!")
finally:
await release_lock_for_file(file)
[Doku]
@classmethod
async def write_binary_file(cls, file: Path, content: bytes) -> Optional[bool]:
"""
Writes binary content to a file (asynchronous).
:param file: The path to the file to write.
:type file: Path
:param content: The binary content to write.
:type content: bytes
:returns: True if writing finished successfully, else False.
:rtype: bool
"""
try:
lock: Lock = await get_lock_for_file(file)
async with lock:
writer_cm = await cls._resolve_async_open(AsyncPath(file).open(mode='wb'))
async with writer_cm as binary_file:
await binary_file.write(content)
return True
except (FileNotFoundError, IOError) as error:
if isinstance(error, FileNotFoundError):
raise FileNotFoundError("File not found!")
elif isinstance(error, IOError):
raise IOError("An unexpected io error occured.")
finally:
await release_lock_for_file(file)
[Doku]
@classmethod
async def write_yaml_file(cls, file: Path, file_content: dict) -> Optional[bool]:
"""
Writes a YAML file from a dictionary (asynchronous).
:param file: The path to the file to write.
:type file: Path
:param file_content: The dictionary to serialize as YAML.
:type file_content: dict
:returns: True if writing finished successfully, else False.
:rtype: bool
"""
try:
lock: Lock = await get_lock_for_file(file)
async with lock:
writer_cm = await cls._resolve_async_open(AsyncPath(file).open(mode='w'))
async with writer_cm as writer:
await writer.write(yaml.dump(file_content))
return True
except (IOError, UnicodeDecodeError, yaml.YAMLError,
TypeError, FileNotFoundError) as error:
if isinstance(error, IOError):
raise IOError('An unexpected io error occured!')
if isinstance(error, UnicodeDecodeError):
raise UnicodeDecodeError('A decoding error occured!', b'', 0, 0, '')
if isinstance(error, yaml.YAMLError):
raise yaml.YAMLError('YAML error occured:', '', 0)
if isinstance(error, TypeError):
raise TypeError("An unexpected type error uccured!")
if isinstance(error, FileNotFoundError):
raise FileNotFoundError("File not found error!")
finally:
await release_lock_for_file(file)
# ========================================================================
# SYNC METHODS
# ========================================================================
[Doku]
@classmethod
def write_json_file_sync(cls, file: Path, file_content: dict) -> Optional[bool]:
"""
Writes a JSON file from a dictionary (synchronous).
:param file: The path to the file to write.
:type file: Path
:param file_content: The dictionary to serialize as JSON.
:type file_content: dict
:returns: True if writing finished successfully, else False.
:rtype: bool
"""
try:
lock = get_lock_for_file_sync(file)
with lock:
with open(file, mode='w', encoding='utf-8') as writer:
writer.write(json.dumps(file_content, indent=4))
return True
except (IOError, UnicodeDecodeError, json.decoder.JSONDecodeError,
TypeError, FileNotFoundError) as error:
if isinstance(error, IOError):
raise IOError('An unexpected io error occured!')
if isinstance(error, UnicodeDecodeError):
raise UnicodeDecodeError('A decoding error occured!', b'', 0, 0, '')
if isinstance(error, json.decoder.JSONDecodeError):
raise json.decoder.JSONDecodeError('JSON decoding error:', '', 0)
if isinstance(error, TypeError):
raise TypeError("An unexpected type error uccured!")
if isinstance(error, FileNotFoundError):
raise FileNotFoundError("File not found error!")
finally:
release_lock_for_file_sync(file)
[Doku]
@classmethod
def write_binary_file_sync(cls, file: Path, content: bytes) -> Optional[bool]:
"""
Writes binary content to a file (synchronous).
:param file: The path to the file to write.
:type file: Path
:param content: The binary content to write.
:type content: bytes
:returns: True if writing finished successfully, else False.
:rtype: bool
"""
try:
lock = get_lock_for_file_sync(file)
with lock:
with open(file, mode='wb') as binary_file:
binary_file.write(content)
return True
except (FileNotFoundError, IOError) as error:
if isinstance(error, FileNotFoundError):
raise FileNotFoundError("File not found!")
elif isinstance(error, IOError):
raise IOError("An unexpected io error occured.")
finally:
release_lock_for_file_sync(file)
[Doku]
@classmethod
def write_yaml_file_sync(cls, file: Path, file_content: dict) -> Optional[bool]:
"""
Writes a YAML file from a dictionary (synchronous).
:param file: The path to the file to write.
:type file: Path
:param file_content: The dictionary to serialize as YAML.
:type file_content: dict
:returns: True if writing finished successfully, else False.
:rtype: bool
"""
try:
lock = get_lock_for_file_sync(file)
with lock:
with open(file, mode='w', encoding='utf-8') as writer:
writer.write(yaml.dump(file_content))
return True
except (IOError, UnicodeDecodeError, yaml.YAMLError,
TypeError, FileNotFoundError) as error:
if isinstance(error, IOError):
raise IOError('An unexpected io error occured!')
if isinstance(error, UnicodeDecodeError):
raise UnicodeDecodeError('A decoding error occured!', b'', 0, 0, '')
if isinstance(error, yaml.YAMLError):
raise yaml.YAMLError('YAML error occured:', '', 0)
if isinstance(error, TypeError):
raise TypeError("An unexpected type error uccured!")
if isinstance(error, FileNotFoundError):
raise FileNotFoundError("File not found error!")
finally:
release_lock_for_file_sync(file)