Skip to content

Commit

Permalink
Creation and modification time
Browse files Browse the repository at this point in the history
  • Loading branch information
edgarrmondragon committed Sep 10, 2024
1 parent bf66f7f commit 4dfdd17
Show file tree
Hide file tree
Showing 3 changed files with 77 additions and 9 deletions.
21 changes: 17 additions & 4 deletions singer_sdk/contrib/filesystem/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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")
Expand Down
33 changes: 28 additions & 5 deletions singer_sdk/contrib/filesystem/local.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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.
Expand All @@ -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.
Expand Down
32 changes: 32 additions & 0 deletions tests/contrib/filesystem/test_local.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
from __future__ import annotations

import datetime
import typing as t
import unittest.mock

from singer_sdk.contrib.filesystem import local

Expand Down Expand Up @@ -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."""

Expand Down

0 comments on commit 4dfdd17

Please sign in to comment.