Skip to content

Commit

Permalink
Merge pull request #1086 from samschott/fix-mypy-issues
Browse files Browse the repository at this point in the history
Fix mypy issues
  • Loading branch information
samschott authored Sep 8, 2024
2 parents 1922320 + 96523ed commit 606e6db
Show file tree
Hide file tree
Showing 5 changed files with 41 additions and 31 deletions.
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ urls = { Homepage = "https://maestral.app" }
requires-python = ">=3.8"
dependencies = [
"click>=8.0.2",
"desktop-notifier>=3.3.0",
"desktop-notifier>=5.0.0",
"dropbox>=11.28.0,<13.0",
"fasteners>=0.15",
"keyring>=22",
Expand Down
4 changes: 2 additions & 2 deletions src/maestral/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@
]


def get_dest_path(event: FileSystemEvent) -> str:
def get_dest_path(event: FileSystemEvent) -> str | bytes:
"""
Returns the dest_path of a file system event if present (moved events only)
otherwise returns the src_path (which is also the "destination").
Expand Down Expand Up @@ -425,7 +425,7 @@ def from_file_system_event(
change_time = stat.st_ctime if stat else None
size = stat.st_size if stat else 0
try:
symlink_target = os.readlink(event.src_path)
symlink_target = os.readlink(os.fsdecode(event.src_path))
except OSError:
symlink_target = None

Expand Down
4 changes: 2 additions & 2 deletions src/maestral/notify.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
from typing import Any, Callable, cast

# external imports
from desktop_notifier import Button, DesktopNotifier, Urgency
from desktop_notifier import Button, DesktopNotifier, Icon, Urgency

# local imports
from .config import MaestralConfig
Expand All @@ -36,7 +36,7 @@

_desktop_notifier = DesktopNotifier(
app_name=APP_NAME,
app_icon=APP_ICON_PATH.as_uri(),
app_icon=Icon(path=APP_ICON_PATH),
)


Expand Down
35 changes: 20 additions & 15 deletions src/maestral/sync.py
Original file line number Diff line number Diff line change
Expand Up @@ -1066,14 +1066,16 @@ def remove_node_from_index(self, dbx_path_lower: str) -> None:

# ==== Content hashing =============================================================

def get_local_hash(self, local_path: str) -> str | None:
def get_local_hash(self, local_path: str | bytes) -> str | None:
"""
Computes content hash of a local file.
:param local_path: Absolute path on local drive.
:returns: Content hash to compare with Dropbox's content hash, or 'folder' if
the path points to a directory. ``None`` if there is nothing at the path.
"""
local_path = os.fsdecode(local_path)

try:
stat = os.lstat(local_path)
except (FileNotFoundError, NotADirectoryError):
Expand Down Expand Up @@ -1360,7 +1362,7 @@ def _correct_case_helper(self, dbx_path: str, dbx_path_lower: str) -> str:

return dbx_path_cased

def to_dbx_path(self, local_path: str) -> str:
def to_dbx_path(self, local_path: str | bytes) -> str:
"""
Converts a local path to a path relative to the Dropbox folder. Casing of the
given ``local_path`` will be preserved.
Expand All @@ -1369,15 +1371,15 @@ def to_dbx_path(self, local_path: str) -> str:
:returns: Relative path with respect to Dropbox folder.
:raises ValueError: When the path lies outside the local Dropbox folder.
"""
if not is_equal_or_child(
local_path, self.dropbox_path, self.is_fs_case_sensitive
):
raise ValueError(f'"{local_path}" is not in "{self.dropbox_path}"')
path = os.fsdecode(local_path)

if not is_equal_or_child(path, self.dropbox_path, self.is_fs_case_sensitive):
raise ValueError(f'"{path}" is not in "{self.dropbox_path}"')
return "/" + removeprefix(
local_path, self.dropbox_path, self.is_fs_case_sensitive
path, self.dropbox_path, self.is_fs_case_sensitive
).lstrip("/")

def to_dbx_path_lower(self, local_path: str) -> str:
def to_dbx_path_lower(self, local_path: str | bytes) -> str:
"""
Converts a local path to a path relative to the Dropbox folder. The path will be
normalized as on Dropbox servers (lower case and some additional
Expand Down Expand Up @@ -1414,7 +1416,7 @@ def to_local_path(self, dbx_path: str) -> str:
dbx_path_cased = self.correct_case(dbx_path)
return self.to_local_path_from_cased(dbx_path_cased)

def is_excluded(self, path: str) -> bool:
def is_excluded(self, path: str | bytes) -> bool:
"""
Checks if a file is excluded from sync. Certain file names are always excluded
from syncing, following the Dropbox support article:
Expand All @@ -1429,6 +1431,7 @@ def is_excluded(self, path: str) -> bool:
just a file name. Does not need to be normalized.
:returns: Whether the path is excluded from syncing.
"""
path = os.fsdecode(path)
dirname, basename = osp.split(path)

# Is in excluded files?
Expand Down Expand Up @@ -1996,7 +1999,9 @@ def _clean_local_events(
# from sync.

# mapping of path -> event history
events_for_path: defaultdict[str, list[FileSystemEvent]] = defaultdict(list)
events_for_path: defaultdict[str | bytes, list[FileSystemEvent]] = defaultdict(
list
)

# mapping of source deletion event -> destination creation event
moved_from_to: dict[FileSystemEvent, FileSystemEvent] = {}
Expand Down Expand Up @@ -2119,8 +2124,8 @@ def _clean_local_events(

# 0) Collect all moved and deleted events in sets.

dir_moved_paths: set[tuple[str, str]] = set()
dir_deleted_paths: set[str] = set()
dir_moved_paths: set[tuple[str | bytes, str | bytes]] = set()
dir_deleted_paths: set[str | bytes] = set()

for events in events_for_path.values():
event = events[0]
Expand All @@ -2132,7 +2137,7 @@ def _clean_local_events(
# 1) Combine moved events of folders and their children into one event.

if len(dir_moved_paths) > 0:
child_moved_dst_paths: set[str] = set()
child_moved_dst_paths: set[str | bytes] = set()

# For each event, check if it is a child of a moved event discard it if yes.
for events in events_for_path.values():
Expand All @@ -2151,7 +2156,7 @@ def _clean_local_events(
# 2) Combine deleted events of folders and their children to one event.

if len(dir_deleted_paths) > 0:
child_deleted_paths: set[str] = set()
child_deleted_paths: set[str | bytes] = set()

for events in events_for_path.values():
event = events[0]
Expand Down Expand Up @@ -3737,7 +3742,7 @@ def _apply_case_change(self, event: SyncEvent) -> None:

self._logger.debug('Renamed "%s" to "%s"', local_path_old, event.local_path)

def rescan(self, local_path: str) -> None:
def rescan(self, local_path: str | bytes) -> None:
"""
Forces a rescan of a local path: schedules created events for every folder,
modified events for every file and deleted events for every deleted item
Expand Down
27 changes: 16 additions & 11 deletions src/maestral/utils/path.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@
This module contains functions for common path operations.
"""

from __future__ import annotations

import errno
import fcntl
import itertools
Expand Down Expand Up @@ -36,7 +38,9 @@ def _path_components(path: str) -> List[str]:
# ==== path relationships ==============================================================


def is_child(path: str, parent: str, case_sensitive: bool = True) -> bool:
def is_child(
path: str | bytes, parent: str | bytes, case_sensitive: bool = True
) -> bool:
"""
Checks if ``path`` semantically is inside ``parent``. Neither path needs to
refer to an actual item on the drive. This function is case-sensitive.
Expand All @@ -49,14 +53,19 @@ def is_child(path: str, parent: str, case_sensitive: bool = True) -> bool:
if not case_sensitive:
path = normalize(path)
parent = normalize(parent)
else:
path = os.fsdecode(path)
parent = os.fsdecode(parent)

parent = parent.rstrip(osp.sep) + osp.sep
path = path.rstrip(osp.sep)

return path.startswith(parent)


def is_equal_or_child(path: str, parent: str, case_sensitive: bool = True) -> bool:
def is_equal_or_child(
path: str | bytes, parent: str | bytes, case_sensitive: bool = True
) -> bool:
"""
Checks if ``path`` semantically is inside ``parent`` or equals ``parent``. Neither
path needs to refer to an actual item on the drive. This function is case-sensitive.
Expand All @@ -67,11 +76,7 @@ def is_equal_or_child(path: str, parent: str, case_sensitive: bool = True) -> bo
:returns: ``True`` if ``path`` semantically lies inside ``parent`` or
``path == parent``.
"""
if not case_sensitive:
path = normalize(path)
parent = normalize(parent)

return is_child(path, parent) or path == parent
return is_child(path, parent, case_sensitive) or path == parent


# ==== case sensitivity and normalization ==============================================
Expand All @@ -98,7 +103,7 @@ def normalize_unicode(string: str) -> str:
return unicodedata.normalize("NFC", string)


def normalize(string: str) -> str:
def normalize(path: str | bytes) -> str:
"""
Replicates the path normalization performed by Dropbox servers. This typically only
involves converting the path to lower case, with a few (undocumented) exceptions:
Expand All @@ -121,7 +126,7 @@ def normalize(string: str) -> str:
:param string: Original path.
:returns: Normalized path.
"""
return normalize_case(normalize_unicode(string))
return normalize_case(normalize_unicode(os.fsdecode(path)))


def is_fs_case_sensitive(path: str) -> bool:
Expand Down Expand Up @@ -417,7 +422,7 @@ def move(


def walk(
root: str,
root: str | bytes,
listdir: Callable[[str], Iterable["os.DirEntry[str]"]] = os.scandir,
) -> Iterator[Tuple[str, os.stat_result]]:
"""
Expand All @@ -427,7 +432,7 @@ def walk(
:param listdir: Function to call to get the folder content.
:returns: Iterator over (path, stat) results.
"""
for entry in listdir(root):
for entry in listdir(os.fsdecode(root)):
try:
path = entry.path
stat = entry.stat(follow_symlinks=False)
Expand Down

0 comments on commit 606e6db

Please sign in to comment.