Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix abstract base class definition #33

Merged
merged 2 commits into from
Oct 27, 2022
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
3 changes: 3 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -100,3 +100,6 @@ dmypy.json

# OSX files
.DS_Store

# Pycharm IDEs
.idea/
116 changes: 92 additions & 24 deletions jupyter_server_fileid/manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import stat
import time
import uuid
from abc import ABC, ABCMeta, abstractmethod
from typing import Any, Callable, Dict, Optional

from jupyter_core.paths import jupyter_data_dir
Expand Down Expand Up @@ -38,15 +39,20 @@ def wrapped(self, *args, **kwargs):
return decorator


class BaseFileIdManager(LoggingConfigurable):
class FileIdManagerMeta(ABCMeta, type(LoggingConfigurable)): # type: ignore
pass


class BaseFileIdManager(ABC, LoggingConfigurable, metaclass=FileIdManagerMeta):
"""
Base class for File ID manager implementations. All File ID
managers should inherit from this class.
"""

root_dir = Unicode(
help=("The root directory being served by Jupyter server. Must be an absolute path."),
help="The root directory being served by Jupyter server.",
config=False,
allow_none=True,
)

db_path = Unicode(
Expand All @@ -58,47 +64,100 @@ class BaseFileIdManager(LoggingConfigurable):
config=True,
)

@validate("root_dir", "db_path")
def _validate_abspath_traits(self, proposal):
@validate("db_path")
def _validate_db_path(self, proposal):
if proposal["value"] is None:
raise TraitError(f"FileIdManager : {proposal['trait'].name} must not be None")
raise TraitError(f"BaseFileIdManager : {proposal['trait'].name} must not be None")
if not os.path.isabs(proposal["value"]):
raise TraitError(f"FileIdManager : {proposal['trait'].name} must be an absolute path")
raise TraitError(
f"BaseFileIdManager : {proposal['trait'].name} must be an absolute path"
)
return proposal["value"]

def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)

def _uuid(self) -> str:
@staticmethod
def _uuid() -> str:
return str(uuid.uuid4())

@abstractmethod
def index(self, path: str) -> Optional[str]:
raise NotImplementedError("must be implemented by subclass")
"""Returns the file ID for the file corresponding to `path`.

If `path` is not already indexed, a new file ID will be created and associated
with `path`, otherwise the existing file ID will be returned. Returns None if
`path` does not correspond to an object as determined by the implementation.
"""
pass

@abstractmethod
def get_id(self, path: str) -> Optional[str]:
raise NotImplementedError("must be implemented by subclass")
"""Retrieves the file ID associated with the given file path.

Returns None if the file has not yet been indexed.
"""
pass

@abstractmethod
def get_path(self, id: str) -> Optional[str]:
raise NotImplementedError("must be implemented by subclass")
"""Retrieves the file path associated with the given file ID.

Returns None if the file ID does not exist.
"""
pass

@abstractmethod
def move(self, old_path: str, new_path: str) -> Optional[str]:
raise NotImplementedError("must be implemented by subclass")
"""Emulates file move operations by updating the old file path to the new file path.

If old_path corresponds to a directory (as determined by the implementation), all indexed
file paths prefixed with old_path will have their locations updated and prefixed with new_path.

Returns the file ID if new_path is valid, otherwise None.
"""
pass

@abstractmethod
def copy(self, from_path: str, to_path: str) -> Optional[str]:
raise NotImplementedError("must be implemented by subclass")
"""Emulates file copy operations by copying the entry corresponding to from_path
and inserting an entry corresponding to to_path.

If from_path corresponds to a directory (as determined by the implementation), all indexed
file paths prefixed with from_path will have their entries copying and inserted to entries
corresponding to to_path.

Returns the file ID if to_path is valid, otherwise None.
"""
pass

@abstractmethod
def delete(self, path: str) -> None:
raise NotImplementedError("must be implemented by subclass")
"""Emulates file delete operations by deleting the entry corresponding to path.

def save(self, path: str) -> None:
raise NotImplementedError("must be implemented by subclass")
If path corresponds to a directory (as determined by the implementation), all indexed
file paths will have their entries deleted.

Returns None.
"""
pass

@abstractmethod
def save(self, path: str) -> Optional[str]:
"""Emulates file save operations by inserting the entry corresponding to path.

Entries are inserted when one corresponding to path does not already exist.

Returns the ID corresponding to path or None if path is determined to not be valid.
"""
pass

@abstractmethod
def get_handlers_by_action(self) -> Dict[str, Optional[Callable[[Dict[str, Any]], Any]]]:
"""Returns a dictionary whose keys are contents manager event actions
and whose values are callables invoked upon receipt of an event of the
same action. The callable accepts the body of the event as its only
argument. To ignore an event action, set the value to `None`."""
raise NotImplementedError("must be implemented by subclass")
"""Returns a dictionary mapping contents manager event actions to a handler (callable).

Returns a dictionary whose keys are contents manager event actions and whose values are callables
invoked upon receipt of an event of the same action. The callable accepts the body of the event as
its only argument. To ignore an event action, set the value to `None`.
"""
pass


class ArbitraryFileIdManager(BaseFileIdManager):
Expand Down Expand Up @@ -224,6 +283,16 @@ class LocalFileIdManager(BaseFileIdManager):
config=True,
)

@validate("root_dir")
def _validate_root_dir(self, proposal):
if proposal["value"] is None:
raise TraitError(f"LocalFileIdManager : {proposal['trait'].name} must not be None")
if not os.path.isabs(proposal["value"]):
raise TraitError(
f"LocalFileIdManager : {proposal['trait'].name} must be an absolute path"
)
return proposal["value"]

def __init__(self, *args, **kwargs):
# pass args and kwargs to parent Configurable
super().__init__(*args, **kwargs)
Expand Down Expand Up @@ -718,7 +787,6 @@ def save(self, path):
row = self.con.execute(
"SELECT id FROM Files WHERE ino = ? AND path = ?", (stat_info.ino, path)
).fetchone()

# if no record exists, return early
if row is None:
return
Expand Down
10 changes: 7 additions & 3 deletions tests/test_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -78,10 +78,14 @@ def get_path_nosync(fid_manager, id):


def test_validates_root_dir(fid_db_path):
rel_root_dir = root_dir = os.path.join("some", "rel", "path")
with pytest.raises(TraitError, match="must be an absolute path"):
LocalFileIdManager(root_dir=os.path.join("some", "rel", "path"), db_path=fid_db_path)
with pytest.raises(TraitError, match="must be an absolute path"):
ArbitraryFileIdManager(root_dir=os.path.join("some", "rel", "path"), db_path=fid_db_path)
LocalFileIdManager(root_dir=rel_root_dir, db_path=fid_db_path)
# root_dir can be relative for ArbitraryFileIdManager instances (and None)
afm = ArbitraryFileIdManager(root_dir=rel_root_dir, db_path=fid_db_path)
assert afm.root_dir == rel_root_dir
afm2 = ArbitraryFileIdManager(root_dir=None, db_path=fid_db_path)
assert afm2.root_dir is None


def test_validates_db_path(jp_root_dir):
Expand Down