diff --git a/singer_sdk/contrib/filesystem/base.py b/singer_sdk/contrib/filesystem/base.py index 09a9ddb191..cbf8f93f3f 100644 --- a/singer_sdk/contrib/filesystem/base.py +++ b/singer_sdk/contrib/filesystem/base.py @@ -5,16 +5,15 @@ import abc import typing as t +if t.TYPE_CHECKING: + import datetime + __all__ = ["AbstractDirectory", "AbstractFile", "AbstractFileSystem"] class AbstractFile(abc.ABC): """Abstract class for file operations.""" - @abc.abstractmethod - def read(self, size: int = -1) -> str: - """Read the file contents.""" - def read_text(self) -> str: """Read the entire file as text. @@ -23,6 +22,20 @@ def read_text(self) -> str: """ return self.read() + @abc.abstractmethod + def read(self, size: int = -1) -> str: + """Read the file contents.""" + + @property + def creation_time(self) -> datetime.datetime: + """Get the creation time of the file.""" + raise NotImplementedError + + @property + def modified_time(self) -> datetime.datetime: + """Get the last modified time of the file.""" + raise NotImplementedError + _F = t.TypeVar("_F") _D = t.TypeVar("_D") diff --git a/singer_sdk/contrib/filesystem/local.py b/singer_sdk/contrib/filesystem/local.py index 5abc44272d..2ad9936278 100644 --- a/singer_sdk/contrib/filesystem/local.py +++ b/singer_sdk/contrib/filesystem/local.py @@ -2,8 +2,9 @@ from __future__ import annotations -import pathlib import typing as t +from datetime import datetime +from pathlib import Path from singer_sdk.contrib.filesystem import base @@ -13,10 +14,10 @@ class LocalFile(base.AbstractFile): """Local file operations.""" - def __init__(self, filepath: str | pathlib.Path): + def __init__(self, filepath: str | Path): """Create a new LocalFile instance.""" self._filepath = filepath - self.path = pathlib.Path(self._filepath).absolute() + self.path = Path(self._filepath).absolute() def __repr__(self) -> str: """A string representation of the LocalFile. @@ -38,14 +39,36 @@ def read(self, size: int = -1) -> str: with self.path.open("r") as file: return file.read(size) + @property + def creation_time(self) -> datetime: + """Get the creation time of the file. + + Returns: + The creation time of the file. + """ + stat = self.path.stat() + try: + return datetime.fromtimestamp(stat.st_birthtime).astimezone() + except AttributeError: + return datetime.fromtimestamp(stat.st_ctime).astimezone() + + @property + def modified_time(self) -> datetime: + """Get the last modified time of the file. + + Returns: + The last modified time of the file. + """ + return datetime.fromtimestamp(self.path.stat().st_mtime).astimezone() + class LocalDirectory(base.AbstractDirectory[LocalFile]): """Local directory operations.""" - def __init__(self, dirpath: str | pathlib.Path): + def __init__(self, dirpath: str | Path): """Create a new LocalDirectory instance.""" self._dirpath = dirpath - self.path = pathlib.Path(self._dirpath).absolute() + self.path = Path(self._dirpath).absolute() def __repr__(self) -> str: """A string representation of the LocalDirectory. diff --git a/tests/contrib/filesystem/test_local.py b/tests/contrib/filesystem/test_local.py index a117c67a9a..0a2c574737 100644 --- a/tests/contrib/filesystem/test_local.py +++ b/tests/contrib/filesystem/test_local.py @@ -1,6 +1,8 @@ from __future__ import annotations +import datetime import typing as t +import unittest.mock from singer_sdk.contrib.filesystem import local @@ -30,6 +32,36 @@ def test_file_read(tmp_path: pathlib.Path): assert file.read(3) == "Hel" +def test_file_creation_time(tmp_path: pathlib.Path): + """Test getting the creation time of a file.""" + + path = tmp_path / "test.txt" + path.write_text("Hello, world!") + + file = local.LocalFile(path) + assert isinstance(file.creation_time, datetime.datetime) + + with unittest.mock.patch("pathlib.Path.stat") as mock_stat: + ts = 1704067200 + mock_stat.return_value = unittest.mock.Mock(st_birthtime=ts, st_ctime=ts) + assert file.creation_time.timestamp() == ts + + +def test_file_modified_time(tmp_path: pathlib.Path): + """Test getting the last modified time of a file.""" + + path = tmp_path / "test.txt" + path.write_text("Hello, world!") + + file = local.LocalFile(path) + assert isinstance(file.modified_time, datetime.datetime) + + with unittest.mock.patch("pathlib.Path.stat") as mock_stat: + ts = 1704067200 + mock_stat.return_value = unittest.mock.Mock(st_mtime=ts) + assert file.modified_time.timestamp() == ts + + def test_directory_list_contents(tmp_path: pathlib.Path): """Test listing a directory."""