Skip to content

File Mover

Bases: FrozenModel

Allows you to move files between different directories in a filesystem, and return an object with move result summary. support hooks

Note

This class is used to move files only within the same connection,

It does NOT support direct file transfer between filesystems, like FTP -> SFTP. You should use File Downloader + File Uploader to implement FTP -> local dir -> SFTP.

Warning

This class does not support read strategies.

Added in 0.8.0

Parameters:

  • connection (FileConnection) –

    Class which contains File system connection properties. See File Connections section.

  • target_path (PathLike | str) –

    Remote path to move files to

  • source_path (PathLike | str, default: None ) –

    Remote path to move files from.

    Could be None, but only if you pass absolute file paths directly to run method

  • filters (list[BaseFileFilter]) –

    Return only files/directories matching these filters. See File Filters

  • limits (list[BaseFileLimit]) –

    Apply limits to the list of files/directories, and stop if one of the limits is reached. See File Limits

  • options (Options | dict | None, default: None ) –

    File moving options. See FileMover.Options

Examples:

from onetl.connection import SFTP
from onetl.file import FileMover

sftp = SFTP(...)

# create mover
mover = FileMover(
    connection=sftp,
    source_path="/path/to/source/dir",
    target_path="/path/to/target/dir",
)

# move files from "/path/to/source/dir" to "/path/to/target/dir"
mover.run()

from onetl.connection import SFTP
from onetl.file import FileMover
from onetl.file.filter import Glob, ExcludeDir
from onetl.file.limit import MaxFilesCount, TotalFilesSize

sftp = SFTP(...)

# create mover with a bunch of options
mover = FileMover(
    connection=sftp,
    source_path="/path/to/source/dir",
    target_path="/path/to/target/dir",
    filters=[
        Glob("*.txt"),
        ExcludeDir("/path/to/source/dir/exclude"),
    ],
    limits=[MaxFilesCount(100), TotalFileSize("10GiB")],
    options=FileMover.Options(if_exists="replace_file"),
)

# move files from "/path/to/source/dir" to "/path/to/target/dir",
# but only *.txt files
# excluding files from "/path/to/source/dir/exclude" directory
# and stop before downloading 101 file
mover.run()

run(files=None)

Method for moving files from source to target directory. support hooks

Added in 0.8.0

Parameters:

  • files (Iterable[str | PathLike] | None, default: None ) –

    File list to move.

    If empty, move files from source_path to target_path, applying filter and limit to each one (if set).

    If not, move to target_path all input files, without any filtering and limiting.

Returns:

  • MoveResult –

    Move result object

Raises:

  • DirectoryNotFoundError –

    source_path does not found

  • NotADirectoryError –

    source_path or target_path is not a directory

Examples:

Move files from source_path:

>>> from onetl.file import FileMover
>>> mover = FileMover(source_path="/source", target_path="/target", ...)
>>> move_result = mover.run()
>>> move_result
MoveResult(
    successful=FileSet([
        RemoteFile("/target/file1.txt"),
        RemoteFile("/target/file2.txt"),
        # directory structure is preserved
        RemoteFile("/target/nested/path/file3.txt"),
    ]),
    failed=FileSet([
        FailedRemoteFile("/source/failed.file"),
    ]),
    skipped=FileSet([
        RemoteFile("/source/already.exists"),
    ]),
    missing=FileSet([
        RemotePath("/source/missing.file"),
    ]),
)

Move only certain files from source_path:

>>> from onetl.file import FileMover
>>> mover = FileMover(source_path="/source", target_path="/target", ...)
>>> # paths could be relative or absolute, but all should be in "/source"
>>> move_result = mover.run(
...     [
...         "/source/file1.txt",
...         "/source/nested/path/file3.txt",
...         # excluding "/source/file2.txt"
...     ]
... )
>>> move_result
MoveResult(
    successful=FileSet([
        RemoteFile("/target/file1.txt"),
        # directory structure is preserved
        RemoteFile("/target/nested/path/file3.txt"),
    ]),
    failed=FileSet([]),
    skipped=FileSet([]),
    missing=FileSet([]),
)

Move certain files from any folder:

>>> from onetl.file import FileMover
>>> mover = FileMover(target_path="/target", ...)  # no source_path set
>>> # only absolute paths
>>> move_result = mover.run(
...     [
...         "/remote/file1.txt",
...         "/any/nested/path/file3.txt",
...     ]
... )
>>> move_result
MoveResult(
    successful=FileSet([
        RemoteFile("/target/file1.txt"),
        # directory structure is NOT preserved without source_path
        RemoteFile("/target/file3.txt"),
    ]),
    failed=FileSet([]),
    skipped=FileSet([]),
    missing=FileSet([]),
)

view_files()

Get file list in the source_path, after filter and limit applied (if any). support hooks

Added in 0.8.0

Raises:

  • DirectoryNotFoundError –

    source_path does not found

  • NotADirectoryError –

    source_path is not a directory

Returns:

  • FileSet[RemoteFile] –

    Set of files in source_path, which will be moved by run method

Examples:

View files:

>>> from onetl.file import FileMover
>>> mover = FileMover(source_path="/remote", ...)
>>> mover.view_files()
FileSet([
    RemoteFile("/remote/file1.txt"),
    RemoteFile("/remote/file2.txt"),
    RemoteFile("/remote/nested/path/file3.txt"),
])