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.

Parameters
  • sbuf – Source buffer

  • target_stream – Target stream to write to.

Returns

Function to be called to apply txdelta windows

Read the svn link’s content.

Parameters

data – svn link’s raw content

Returns

The tuple of (filetype, destination path)

Determine if a filepath is an svnlink or something else.

Parameters

fullpath – Full path to the potential symlink to check

Returns

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

Type

keep track of non link file content with svn

executable: int = 0

executable property

Type

keep track if file is executable when setting svn

special property

Type

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.

directory
path
fullpath
state
svnrepo
editor
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)

executable
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.

directory
rootpath
path
file_states
dir_states
svnrepo
editor
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.

Parameters

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.

close()[source]

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.

Returns

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.

Parameters

rev – The revision to start the replay from.

Returns

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