Coverage for enderchest/sync/__init__.py: 100%

1"""Low-level functionality for synchronizing across different machines""" 

2import importlib 

3from contextlib import contextmanager 

4from pathlib import Path 

5from tempfile import TemporaryDirectory 

6from typing import Collection, Generator 

7from urllib.parse import ParseResult 

8 

9from ..loggers import SYNC_LOGGER 

10from .utils import Operation as Op 

11from .utils import ( 

12 abspath_from_uri, 

13 diff, 

14 filter_contents, 

15 generate_sync_report, 

16 get_default_netloc, 

17 is_identical, 

18 render_remote, 

19 uri_to_ssh, 

20) 

21 

22SUPPORTED_PROTOCOLS = ("rsync", "sftp", "file") 

23 

24DEFAULT_PROTOCOL = SUPPORTED_PROTOCOLS[0] 

25 

26 

27def pull( 

28 remote_uri: ParseResult, 

29 local_path: Path, 

30 exclude: Collection[str] | None = None, 

31 dry_run: bool = False, 

32 **kwargs, 

33) -> None: 

34 """Pull all upstream changes from a remote into the specified location 

35 

36 Parameters 

37 ---------- 

38 remote_uri : ParseResult 

39 The URI for the remote resource to pull 

40 local_path : Path 

41 The local destination 

42 exclude : list of str, optional 

43 Any patterns that should be excluded from the sync 

44 dry_run : bool, optional 

45 Whether to only simulate this sync (report the operations to be performed 

46 but not actually perform them). Default is False. 

47 **kwargs 

48 Any additional options to pass into the sync command 

49 """ 

50 try: 

51 protocol = importlib.import_module(f"{__package__}.{remote_uri.scheme.lower()}") 

52 protocol.pull(remote_uri, local_path, exclude or (), dry_run, **kwargs) 

53 except ModuleNotFoundError as not_installed: # pragma: no cover 

54 raise NotImplementedError( 

55 f"Protocol {remote_uri.scheme} is not currently implemented" 

56 ) from not_installed 

57 

58 

59def push( 

60 local_path: Path, 

61 remote_uri: ParseResult, 

62 exclude: Collection[str] | None = None, 

63 dry_run: bool = False, 

64 **kwargs, 

65) -> None: 

66 """Push all local changes in the specified directory into the specified remote 

67 

68 Parameters 

69 ---------- 

70 local_path 

71 The local path to push 

72 remote_uri : ParseResult 

73 The URI for the remote destination 

74 exclude : list of str, optional 

75 Any patterns that should be excluded from the sync 

76 dry_run : bool, optional 

77 Whether to only simulate this sync (report the operations to be performed 

78 but not actually perform them). Default is False. 

79 **kwargs 

80 Any additional options to pass into the sync command 

81 """ 

82 try: 

83 protocol = importlib.import_module(f"{__package__}.{remote_uri.scheme.lower()}") 

84 protocol.push(local_path, remote_uri, exclude or (), dry_run, **kwargs) 

85 except ModuleNotFoundError as not_installed: 

86 raise NotImplementedError( 

87 f"Protocol {remote_uri.scheme} is not currently implemented" 

88 ) from not_installed 

89 

90 

91@contextmanager 

92def remote_file(uri: ParseResult) -> Generator[Path, None, None]: 

93 """Grab a file from a remote filesystem by its URI and read its contents 

94 

95 Parameters 

96 ---------- 

97 uri : parsed URI 

98 The URI of the file to read 

99 

100 Yields 

101 ------ 

102 Path 

103 A path to a local (temp) copy of the file 

104 """ 

105 with TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir: 

106 pull(uri, Path(tmpdir), verbosity=-1) 

107 yield Path(tmpdir) / Path(uri.path).name 

108 

109 

110__all__ = [ 

111 "SYNC_LOGGER", 

112 "SUPPORTED_PROTOCOLS", 

113 "DEFAULT_PROTOCOL", 

114 "get_default_netloc", 

115 "render_remote", 

116 "remote_file", 

117 "abspath_from_uri", 

118 "pull", 

119 "push", 

120 "uri_to_ssh", 

121]