Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/svn/replay.py
| Show First 20 Lines • Show All 437 Lines • ▼ Show 20 Lines | def open_directory(self, path: str, *args) -> DirEditor: | ||||
| self.directory, | self.directory, | ||||
| rootpath=self.rootpath, | rootpath=self.rootpath, | ||||
| path=os.fsencode(path), | path=os.fsencode(path), | ||||
| file_states=self.file_states, | file_states=self.file_states, | ||||
| dir_states=self.dir_states, | dir_states=self.dir_states, | ||||
| svnrepo=self.svnrepo, | svnrepo=self.svnrepo, | ||||
| ) | ) | ||||
| def add_directory(self, path: str, *args) -> DirEditor: | def add_directory( | ||||
| self, path: str, copyfrom_path: Optional[str] = None, copyfrom_rev: int = -1 | |||||
| ) -> DirEditor: | |||||
| """Adding a new directory.""" | """Adding a new directory.""" | ||||
| path_bytes = os.fsencode(path) | path_bytes = os.fsencode(path) | ||||
| fullpath = os.path.join(self.rootpath, path_bytes) | |||||
| os.makedirs(os.path.join(self.rootpath, path_bytes), exist_ok=True) | os.makedirs(fullpath, exist_ok=True) | ||||
| if copyfrom_rev == -1: | |||||
| if path_bytes and path_bytes not in self.directory: | if path_bytes and path_bytes not in self.directory: | ||||
| self.dir_states[path_bytes] = DirState() | self.dir_states[path_bytes] = DirState() | ||||
| self.directory[path_bytes] = from_disk.Directory() | self.directory[path_bytes] = from_disk.Directory() | ||||
| else: | |||||
| url = svn_urljoin(self.svnrepo.remote_url, copyfrom_path) | |||||
| self.remove_child(path_bytes) | |||||
| self.svnrepo.export( | |||||
| url, | |||||
| to=fullpath, | |||||
| peg_rev=copyfrom_rev, | |||||
| ignore_keywords=True, | |||||
| overwrite=True, | |||||
| ignore_externals=True, | |||||
| ) | |||||
| self.directory[path_bytes] = from_disk.Directory.from_disk(path=fullpath) | |||||
| return DirEditor( | return DirEditor( | ||||
| self.directory, | self.directory, | ||||
| self.rootpath, | self.rootpath, | ||||
| path_bytes, | path_bytes, | ||||
| self.file_states, | self.file_states, | ||||
| self.dir_states, | self.dir_states, | ||||
| svnrepo=self.svnrepo, | svnrepo=self.svnrepo, | ||||
| ) | ) | ||||
| def open_file(self, path: str, *args) -> FileEditor: | def open_file(self, path: str, *args) -> FileEditor: | ||||
| """Updating existing file.""" | """Updating existing file.""" | ||||
| path_bytes = os.fsencode(path) | path_bytes = os.fsencode(path) | ||||
| self.directory[path_bytes] = from_disk.Content() | self.directory[path_bytes] = from_disk.Content() | ||||
| fullpath = os.path.join(self.rootpath, path_bytes) | fullpath = os.path.join(self.rootpath, path_bytes) | ||||
| return FileEditor( | return FileEditor( | ||||
| self.directory, | self.directory, | ||||
| rootpath=self.rootpath, | rootpath=self.rootpath, | ||||
| path=path_bytes, | path=path_bytes, | ||||
| state=self.file_states[fullpath], | state=self.file_states[fullpath], | ||||
| svnrepo=self.svnrepo, | svnrepo=self.svnrepo, | ||||
| ) | ) | ||||
| def add_file(self, path: str, *args) -> FileEditor: | def add_file( | ||||
| self, path: str, copyfrom_path: Optional[str] = None, copyfrom_rev: int = -1 | |||||
| ) -> FileEditor: | |||||
| """Creating a new file.""" | """Creating a new file.""" | ||||
| path_bytes = os.fsencode(path) | path_bytes = os.fsencode(path) | ||||
| self.directory[path_bytes] = from_disk.Content() | |||||
| fullpath = os.path.join(self.rootpath, path_bytes) | fullpath = os.path.join(self.rootpath, path_bytes) | ||||
| self.file_states[fullpath] = FileState() | self.file_states[fullpath] = FileState() | ||||
| if copyfrom_rev == -1: | |||||
| self.directory[path_bytes] = from_disk.Content() | |||||
| else: | |||||
| url = svn_urljoin(self.svnrepo.remote_url, copyfrom_path) | |||||
| self.remove_child(path_bytes) | |||||
| self.svnrepo.export( | |||||
| url, | |||||
| to=fullpath, | |||||
| peg_rev=copyfrom_rev, | |||||
| ignore_keywords=True, | |||||
| overwrite=True, | |||||
| ) | |||||
| self.directory[path_bytes] = from_disk.Content.from_file(path=fullpath) | |||||
| return FileEditor( | return FileEditor( | ||||
| self.directory, | self.directory, | ||||
| self.rootpath, | self.rootpath, | ||||
| path_bytes, | path_bytes, | ||||
| state=self.file_states[fullpath], | state=self.file_states[fullpath], | ||||
| svnrepo=self.svnrepo, | svnrepo=self.svnrepo, | ||||
| ) | ) | ||||
| ▲ Show 20 Lines • Show All 435 Lines • ▼ Show 20 Lines | ): | ||||
| self.rootpath = rootpath | self.rootpath = rootpath | ||||
| if directory is None: | if directory is None: | ||||
| directory = from_disk.Directory() | directory = from_disk.Directory() | ||||
| self.directory = directory | self.directory = directory | ||||
| self.editor = Editor( | self.editor = Editor( | ||||
| rootpath=rootpath, directory=directory, svnrepo=svnrepo, temp_dir=temp_dir | rootpath=rootpath, directory=directory, svnrepo=svnrepo, temp_dir=temp_dir | ||||
| ) | ) | ||||
| def replay(self, rev: int) -> from_disk.Directory: | def replay(self, rev: int, low_water_mark: int) -> from_disk.Directory: | ||||
| """Replay svn actions between rev and rev+1. | """Replay svn actions between rev and rev+1. | ||||
| This method updates in place the self.editor.directory, as well as the | This method updates in place the self.editor.directory, as well as the | ||||
| filesystem. | filesystem. | ||||
| Returns: | Returns: | ||||
| The updated root directory | The updated root directory | ||||
| """ | """ | ||||
| codecs.register_error("strict", _ra_codecs_error_handler) | codecs.register_error("strict", _ra_codecs_error_handler) | ||||
| self.conn.replay(rev, rev + 1, self.editor) | self.conn.replay(rev, low_water_mark, self.editor) | ||||
| codecs.register_error("strict", codecs.strict_errors) | codecs.register_error("strict", codecs.strict_errors) | ||||
| return self.editor.directory | return self.editor.directory | ||||
| def compute_objects( | def compute_objects( | ||||
| self, rev: int | self, rev: int, low_water_mark: int | ||||
| ) -> Tuple[List[Content], List[SkippedContent], List[Directory]]: | ) -> Tuple[List[Content], List[SkippedContent], List[Directory]]: | ||||
| """Compute objects added or modified at revisions rev. | """Compute objects added or modified at revisions rev. | ||||
| Expects the state to be at previous revision's objects. | Expects the state to be at previous revision's objects. | ||||
| Args: | Args: | ||||
| rev: The revision to start the replay from. | rev: The revision to start the replay from. | ||||
| Returns: | Returns: | ||||
| The updated objects between rev and rev+1. Beware that this | The updated objects between rev and rev+1. Beware that this | ||||
| mutates the filesystem at rootpath accordingly. | mutates the filesystem at rootpath accordingly. | ||||
| """ | """ | ||||
| self.replay(rev) | self.replay(rev, low_water_mark) | ||||
| contents: List[Content] = [] | contents: List[Content] = [] | ||||
| skipped_contents: List[SkippedContent] = [] | skipped_contents: List[SkippedContent] = [] | ||||
| directories: List[Directory] = [] | directories: List[Directory] = [] | ||||
| for obj_node in self.directory.collect(): | for obj_node in self.directory.collect(): | ||||
| obj = obj_node.to_model() # type: ignore | obj = obj_node.to_model() # type: ignore | ||||
| obj_type = obj.object_type | obj_type = obj.object_type | ||||
| ▲ Show 20 Lines • Show All 85 Lines • Show Last 20 Lines | |||||