Use pathlib in mappers and writers

This commit is contained in:
Joakim Soderlund 2019-03-09 21:39:56 +01:00
parent 2ad9d09515
commit 82dd078020
4 changed files with 52 additions and 44 deletions

View file

@ -5,7 +5,7 @@ Mappers for Fimfarchive.
#
# Fimfarchive, preserves stories from Fimfiction.
# Copyright (C) 2018 Joakim Soderlund
# Copyright (C) 2019 Joakim Soderlund
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
@ -22,10 +22,10 @@ Mappers for Fimfarchive.
#
import os
import string
from abc import abstractmethod
from html import unescape
from pathlib import Path
from typing import Dict, Generic, Optional, Set, TypeVar, Union
from arrow import api as arrow, Arrow
@ -107,19 +107,22 @@ class StoryDateMapper(Mapper[Optional[Arrow]]):
return None
class StoryPathMapper(Mapper[str]):
class StoryPathMapper(Mapper[Path]):
"""
Returns a key-based file path for a story.
"""
def __init__(self, directory: str) -> None:
self.directory = directory
def __init__(self, directory: Union[Path, str]) -> None:
"""
Constructor.
def __call__(self, story: Story) -> str:
directory = str(self.directory)
key = str(story.key)
Args:
directory: The directory for the path.
"""
self.directory = Path(directory)
return os.path.join(directory, key)
def __call__(self, story: Story) -> Path:
return self.directory / str(story.key)
class StorySlugMapper(Mapper[str]):

View file

@ -23,8 +23,8 @@ Writers for Fimfarchive.
import json
import os
from typing import Callable, Optional, Union
from pathlib import Path
from typing import Callable, Union
from fimfarchive.mappers import StaticMapper, StoryPathMapper
from fimfarchive.stories import Story
@ -36,8 +36,8 @@ __all__ = (
)
PathFunc = Callable[[Story], Optional[str]]
PathSpec = Union[None, PathFunc, str]
PathFunc = Callable[[Story], Union[None, Path, str]]
PathSpec = Union[None, Path, PathFunc, str]
class Writer():
@ -94,14 +94,14 @@ class DirectoryWriter(Writer):
"""
if callable(obj):
return obj
elif isinstance(obj, str):
elif isinstance(obj, (Path, str)):
return StoryPathMapper(obj)
elif obj is None:
return StaticMapper(obj)
else:
raise TypeError("Path must be callable or string.")
def check_overwrite(self, path: str) -> None:
def check_overwrite(self, path: Path) -> None:
"""
Checks that a file is not overwritten unless requested.
@ -111,10 +111,10 @@ class DirectoryWriter(Writer):
Raises:
FileExistsError: If overwrite is disabled and path exists.
"""
if not self.overwrite and os.path.exists(path):
if not self.overwrite and path.exists():
raise FileExistsError("Would overwrite: '{}'." .format(path))
def check_directory(self, path: str) -> None:
def check_directory(self, path: Path) -> None:
"""
Checks that the path's parent directory exists.
@ -128,16 +128,16 @@ class DirectoryWriter(Writer):
FileNotFoundError: If the parent directory does not exist,
and if directory creation has been disabled.
"""
parent = os.path.dirname(path)
parent = path.parent
if os.path.isdir(parent):
if parent.is_dir():
return
elif self.make_dirs:
os.makedirs(parent)
parent.mkdir(parents=True)
else:
raise FileNotFoundError(parent)
def perform_write(self, contents: bytes, path: str) -> None:
def perform_write(self, contents: bytes, path: Path) -> None:
"""
Performs the actual file write.
@ -148,10 +148,9 @@ class DirectoryWriter(Writer):
self.check_overwrite(path)
self.check_directory(path)
with open(path, 'wb') as fobj:
fobj.write(contents)
path.write_bytes(contents)
def write_meta(self, story: Story, path: str) -> None:
def write_meta(self, story: Story, path: Path) -> None:
"""
Prepares the story meta for writing.
@ -169,7 +168,7 @@ class DirectoryWriter(Writer):
contents = text.encode('utf-8')
self.perform_write(contents, path)
def write_data(self, story: Story, path: str) -> None:
def write_data(self, story: Story, path: Path) -> None:
"""
Prepares the story data for writing.
@ -181,11 +180,13 @@ class DirectoryWriter(Writer):
self.perform_write(contents, path)
def write(self, story: Story) -> None:
meta_path = self.meta_path(story)
data_path = self.data_path(story)
meta_target = self.meta_path(story)
data_target = self.data_path(story)
if meta_path:
if meta_target is not None:
meta_path = Path(meta_target).resolve()
self.write_meta(story, meta_path)
if data_path:
if data_target is not None:
data_path = Path(data_target).resolve()
self.write_data(story, data_path)

View file

@ -23,7 +23,8 @@ Mapper tests.
import os
from typing import no_type_check, Dict, Any
from pathlib import Path
from typing import Any, Dict
from unittest.mock import patch, MagicMock, PropertyMock
import pytest
@ -262,23 +263,18 @@ class TestStoryPathMapper:
mapper = StoryPathMapper(directory)
assert mapper(story) == path
assert mapper(story) == Path(path)
@no_type_check
def test_casts_values(self, tmpdir, story):
"""
Tests casts all values to string when joining.
"""
directory = MagicMock()
directory.__str__.return_value = 'dir'
mapper = StoryPathMapper('dir')
story.key = MagicMock()
story.key.__str__.return_value = 'key'
mapper = StoryPathMapper(directory)
assert mapper(story) == os.path.join('dir', 'key')
assert directory.__str__.called_once_with()
assert mapper(story) == Path('dir', 'key')
assert story.key.__str__.called_once_with()

View file

@ -5,7 +5,7 @@ Writer tests.
#
# Fimfarchive, preserves stories from Fimfiction.
# Copyright (C) 2015 Joakim Soderlund
# Copyright (C) 2019 Joakim Soderlund
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
@ -24,6 +24,7 @@ Writer tests.
import json
import os
from pathlib import Path
import pytest
@ -77,7 +78,7 @@ class TestDirectoryWriter:
Tests `TypeError` is raised for invalid path types.
"""
with pytest.raises(TypeError):
DirectoryWriter(meta_path=1)
DirectoryWriter(meta_path=1) # type: ignore
def test_parent_directory_creation(self, story, tmpdir):
"""
@ -146,8 +147,15 @@ class TestDirectoryWriter:
writer.write(story)
with open(meta_path(story), 'rt') as fobj:
assert story.meta == json.load(fobj)
with open(meta_path(story), 'rt') as meta_stream:
assert story.meta == json.load(meta_stream)
with open(data_path(story), 'rb') as fobj:
assert story.data == fobj.read()
with open(data_path(story), 'rb') as data_stream:
assert story.data == data_stream.read()
def test_current_directory_check(self, story):
"""
Tests directory check for current directory.
"""
writer = DirectoryWriter()
writer.check_directory(Path('key'))