swh.loader.svn.replay module

Remote Access client to svn server.

swh.loader.svn.replay.apply_txdelta_handler(sbuf: bytes, target_stream: BinaryIO) Callable[[Any, bytes, BinaryIO], None][source]

Return a function that can be called repeatedly with txdelta windows. When done, closes the target_stream.

Adapted from subvertpy.delta.apply_txdelta_handler to close the stream when done.

  • sbuf – Source buffer

  • target_stream – Target stream to write to.


Function to be called to apply txdelta windows

Read the svn link’s content.


data – svn link’s raw content


The tuple of (filetype, destination path)

Determine if a filepath is an svnlink or something else.


fullpath – Full path to the potential symlink to check


Tuple containing a boolean value to determine if it’s indeed a symlink (as per svn) and the link target.

class swh.loader.svn.replay.FileState(eol_style: Optional[str] = None, svn_special_path_non_link_data: Optional[bytes] = None, executable: int = 0, link: bool = False)[source]

Bases: object

Persists some file states (eg. end of lines style) across revisions while replaying them.

eol_style: Optional[str] = None

EOL state check mess

special property set


keep track of non link file content with svn

executable: int = 0

executable property


keep track if file is executable when setting svn

special property


keep track if file is a svn link when setting svn

class swh.loader.svn.replay.FileEditor(directory: from_disk.Directory, rootpath: bytes, path: bytes, state: FileState, svnrepo: SvnRepo)[source]

Bases: object

File Editor in charge of updating file on disk and memory objects.

change_prop(key: str, value: str) None[source]
apply_textdelta(base_checksum) Callable[[Any, bytes, BinaryIO], None][source]
close() None[source]

When done with the file, this is called.

So the file exists and is updated, we can:

  • adapt accordingly its execution flag if any

  • compute the objects’ checksums

  • replace the svnlink with a real symlink (for disk computation purposes)

class swh.loader.svn.replay.DirState(externals: Dict[str, List[ExternalDefinition]] = <factory>)[source]

Bases: object

Persists some directory states (eg. externals) across revisions while replaying them.

externals: Dict[str, List[Tuple[str, Optional[int], bool]]]

Map a path in the directory to a list of (external_url, revision, relative_url) targeting it

class swh.loader.svn.replay.DirEditor(directory: from_disk.Directory, rootpath: bytes, path: bytes, file_states: Dict[bytes, FileState], dir_states: Dict[bytes, DirState], svnrepo: SvnRepo)[source]

Bases: object

Directory Editor in charge of updating directory hashes computation.

This implementation includes empty folder in the hash computation.

externals: Dict[str, List[Tuple[str, Optional[int], bool]]]
remove_child(path: bytes) None[source]

Remove a path from the current objects.

The path can be resolved as link, file or directory.

This function takes also care of removing the link between the child and the parent.


path – to remove from the current objects.

open_directory(path: str, *args) swh.loader.svn.replay.DirEditor[source]

Updating existing directory.

add_directory(path: str, *args) swh.loader.svn.replay.DirEditor[source]

Adding a new directory.

open_file(path: str, *args) swh.loader.svn.replay.FileEditor[source]

Updating existing file.

add_file(path: str, *args) swh.loader.svn.replay.FileEditor[source]

Creating a new file.

change_prop(key: str, value: str) None[source]

Change property callback on directory.

delete_entry(path: str, revision: int) None[source]

Remove a path.


Function called when we finish processing a repository.

SVN external definitions are processed by it.

process_external(path: str, external_url: str, revision: Optional[int], relative_url: bool, remove_target_path: bool = True) None[source]
remove_external_path(external_path: bytes, remove_subpaths: bool = True, force: bool = False) None[source]

Remove a previously exported SVN external path from the reconstructed filesystem.

class swh.loader.svn.replay.Editor(rootpath: bytes, directory: from_disk.Directory, svnrepo: SvnRepo, temp_dir: str)[source]

Bases: object

Editor in charge of replaying svn events and computing objects along.

This implementation accounts for empty folder during hash computations.

set_target_revision(revnum) None[source]
abort() None[source]
close() None[source]
open_root(base_revnum: int) swh.loader.svn.replay.DirEditor[source]
class swh.loader.svn.replay.Replay(conn: RemoteAccess, rootpath: bytes, svnrepo: SvnRepo, temp_dir: str, directory: Optional[from_disk.Directory] = None)[source]

Bases: object

Replay class.

replay(rev: int) swh.model.from_disk.Directory[source]

Replay svn actions between rev and rev+1.

This method updates in place the self.editor.directory, as well as the filesystem.


The updated root directory

compute_objects(rev: int) Tuple[List[swh.model.model.Content], List[swh.model.model.SkippedContent], List[swh.model.model.Directory]][source]

Compute objects added or modified at revisions rev. Expects the state to be at previous revision’s objects.


rev – The revision to start the replay from.


The updated objects between rev and rev+1. Beware that this mutates the filesystem at rootpath accordingly.