Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
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
73 changes: 73 additions & 0 deletions src/project/common/utils/file/base.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,73 @@
from pathlib import Path
from typing import Any, Protocol, TypeVar

T = TypeVar('T')

JsonLikeValue = dict[str, Any] | list[Any] | str | int | float | bool | None


class FileLoader(Protocol):
"""Protocol for loading data from files.

Implementations should handle file format-specific deserialization.
"""

def load(self, path: str | Path) -> Any: # noqa: ANN401
"""Load data from the specified file path.

Args:
path: Path to the file to load

Returns:
Deserialized data from the file

"""
...


class FileSaver(Protocol):
"""Protocol for saving data to files.

Implementations should handle file format-specific serialization.
"""

def save(
self,
data: Any, # noqa: ANN401
path: str | Path,
*,
parents: bool = True,
exist_ok: bool = True,
) -> None:
"""Save data to the specified file path.

Args:
data: Data to serialize and save
path: Path where the file should be saved
parents: If True, create parent directories as needed
exist_ok: If True, don't raise error if directory exists

"""
...


class FileHandler(Protocol):
"""Combined protocol for both loading and saving files.

Provides a complete interface for file I/O operations.
"""

def load(self, path: str | Path) -> Any: # noqa: ANN401
"""Load data from the specified file path."""
...

def save(
self,
data: Any, # noqa: ANN401
path: str | Path,
*,
parents: bool = True,
exist_ok: bool = True,
) -> None:
"""Save data to the specified file path."""
...
29 changes: 15 additions & 14 deletions src/project/common/utils/file/config.py
Original file line number Diff line number Diff line change
@@ -1,23 +1,24 @@
from pathlib import Path
from typing import Any

from project.common.utils.file.json import load_json
from project.common.utils.file.toml import load_toml
from project.common.utils.file.yaml import load_yaml
from project.common.utils.file.io import load_file


def load_config(path: str | Path) -> dict[str, Any]:
"""Load configuration from a file (JSON, YAML, or TOML)."""
ext = Path(path).suffix.lower()

if ext == '.json':
data = load_json(path)
elif ext in ('.yaml', '.yml'):
data = load_yaml(path)
elif ext == '.toml':
data = load_toml(path)
else:
raise ValueError(f'Unsupported config file format: {ext}')
"""Load configuration from a file (JSON, YAML, TOML, XML).

Args:
path: Path to the configuration file. Format is detected from extension.

Returns:
Configuration data as a dictionary.

Raises:
ValueError: If file format is not supported.
TypeError: If the loaded data is not a dictionary.

"""
data = load_file(path)

if not isinstance(data, dict):
raise TypeError(f'Config file {path!r} did not return a dict, got {type(data).__name__}')
Expand Down
91 changes: 91 additions & 0 deletions src/project/common/utils/file/factory.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,91 @@
from pathlib import Path
from typing import ClassVar, Literal

from project.common.utils.file.base import FileHandler
from project.common.utils.file.json import JsonFileHandler
from project.common.utils.file.toml import TomlFileHandler
from project.common.utils.file.xml import XmlFileHandler
from project.common.utils.file.yaml import YamlFileHandler

FileFormat = Literal['json', 'yaml', 'toml', 'xml']


class FileHandlerFactory:
"""Factory for creating file handlers based on file format."""

_handlers: ClassVar[dict[FileFormat, type[FileHandler]]] = {
'json': JsonFileHandler,
'yaml': YamlFileHandler,
'toml': TomlFileHandler,
'xml': XmlFileHandler,
}

@classmethod
def create(cls, format_type: FileFormat) -> FileHandler:
"""Create a file handler for the specified format.

Args:
format_type: File format ('json', 'yaml', or 'toml')

Returns:
File handler instance for the specified format

Raises:
ValueError: If format_type is not supported

"""
handler_class = cls._handlers.get(format_type)
if handler_class is None:
supported = ', '.join(cls._handlers.keys())
msg = f'Unsupported file format: {format_type}. Supported formats: {supported}'
raise ValueError(msg)
return handler_class()

@classmethod
def from_path(cls, path: str | Path) -> FileHandler:
"""Create a file handler by detecting format from file extension.

Args:
path: File path with extension

Returns:
File handler instance for the detected format

Raises:
ValueError: If file extension is not recognized or missing

"""
suffix = Path(path).suffix.lstrip('.')
if not suffix:
msg = f'Cannot detect file format: no extension in {path}'
raise ValueError(msg)

# Map common extensions to format types
extension_map: dict[str, FileFormat] = {
'json': 'json',
'yaml': 'yaml',
'yml': 'yaml',
'toml': 'toml',
'xml': 'xml',
}

format_type = extension_map.get(suffix.lower())
if format_type is None:
supported = ', '.join(extension_map.keys())
msg = f'Unsupported file extension: .{suffix}. Supported extensions: {supported}'
raise ValueError(msg)

return cls.create(format_type)


def get_file_handler(path: str | Path) -> FileHandler:
"""Get a file handler from a file path.

Args:
path: File path with extension

Returns:
File handler instance for the detected format

"""
return FileHandlerFactory.from_path(path)
60 changes: 60 additions & 0 deletions src/project/common/utils/file/io.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,60 @@
"""Generic file I/O operations using FileHandler abstraction.

This module provides format-agnostic file operations that automatically
detect and handle different file formats (JSON, YAML, TOML).
"""

from pathlib import Path
from typing import Any

from project.common.utils.file.factory import get_file_handler


def load_file(path: str | Path) -> Any: # noqa: ANN401
"""Load data from a file, automatically detecting format from extension.

Args:
path: Path to the file (extension determines format)

Returns:
Deserialized data from the file

Raises:
ValueError: If file format cannot be detected or is unsupported

Example:
>>> data = load_file('config.json')
>>> data = load_file('settings.yaml')
>>> data = load_file('pyproject.toml')

"""
handler = get_file_handler(path)
return handler.load(path)


def save_file(
data: Any, # noqa: ANN401
path: str | Path,
*,
parents: bool = True,
exist_ok: bool = True,
) -> None:
"""Save data to a file, automatically detecting format from extension.

Args:
data: Data to save
path: Path where the file should be saved (extension determines format)
parents: If True, create parent directories as needed
exist_ok: If True, don't raise error if directory exists

Raises:
ValueError: If file format cannot be detected or is unsupported

Example:
>>> save_file({'key': 'value'}, 'output.json')
>>> save_file(['item1', 'item2'], 'output.yaml')
>>> save_file({'tool': {'poetry': {}}}, 'pyproject.toml')

"""
handler = get_file_handler(path)
handler.save(data, path, parents=parents, exist_ok=exist_ok)
19 changes: 19 additions & 0 deletions src/project/common/utils/file/json.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,3 +20,22 @@ def save_as_indented_json(
target.parent.mkdir(parents=parents, exist_ok=exist_ok)
with target.open(mode='w', encoding='utf-8') as fout:
json.dump(data, fout, ensure_ascii=False, indent=4, separators=(',', ': '))


class JsonFileHandler:
"""JSON file handler implementing FileHandler protocol."""

def load(self, path: str | Path) -> JsonValue:
"""Load JSON data from file."""
return load_json(path)

def save(
self,
data: JsonValue,
path: str | Path,
*,
parents: bool = True,
exist_ok: bool = True,
) -> None:
"""Save data as indented JSON to file."""
save_as_indented_json(data, path, parents=parents, exist_ok=exist_ok)
19 changes: 19 additions & 0 deletions src/project/common/utils/file/toml.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,3 +19,22 @@ def save_as_toml(
target.parent.mkdir(parents=parents, exist_ok=exist_ok)
with target.open(mode='w', encoding='utf-8') as fout:
toml.dump(data, fout)


class TomlFileHandler:
"""TOML file handler implementing FileHandler protocol."""

def load(self, path: str | Path) -> dict[str, Any]:
"""Load TOML data from file."""
return load_toml(path)

def save(
self,
data: dict[str, Any],
path: str | Path,
*,
parents: bool = True,
exist_ok: bool = True,
) -> None:
"""Save data as TOML to file."""
save_as_toml(data, path, parents=parents, exist_ok=exist_ok)
Loading