File Mover

FileMover	Allows you to move files between different directories in a filesystem, and return an object with move result summary.
FileMover.run([files])	Method for moving files from source to target directory.
FileMover.view_files()	Get file list in the source_path, after filter and limit applied (if any).

class onetl.file.file_mover.file_mover.FileMover(*, connection: ~onetl.base.base_file_connection.BaseFileConnection, target_path: ~onetl.impl.remote_path.RemotePath, source_path: ~onetl.impl.remote_path.RemotePath | None = None, filters: ~typing.List[~onetl.base.base_file_filter.BaseFileFilter] = None, limits: ~typing.List[~onetl.base.base_file_limit.BaseFileLimit] = None, options: ~onetl.file.file_mover.options.FileMoverOptions = FileMoverOptions(if_exists=<FileExistBehavior.ERROR: 'error'>, workers=1))

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

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.

Parameters:

connectiononetl.connection.FileConnection

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

target_pathos.PathLike or str

Remote path to move files to

source_pathos.PathLike or str, optional, default: None

Remote path to move files from.

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

filterslist of BaseFileFilter

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

limitslist of BaseFileLimit

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

optionsOptions | dict | None, default: None

File moving options. See Options

Examples

Simple Mover creation

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()

Mover with all parameters

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

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)],
    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: Iterable[str | PathLike] | None = None) → MoveResult

Method for moving files from source to target directory.

Parameters:

filesIterable[str | os.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:

moved_filesMoveResult

Move result object

Raises:

onetl.exception.DirectoryNotFoundError

source_path does not found

NotADirectoryError

source_path or target_path is not a directory

Examples

Move files from source_path

from onetl.impl import RemoteFile, RemotePath
from onetl.file import FileMover

mover = FileMover(source_path="/source", target_path="/target", ...)
moved_files = mover.run()

assert moved_files.successful == {
    RemoteFile("/target/file1.txt"),
    RemoteFile("/target/file2.txt"),
    RemoteFile("/target/nested/path/file3.txt"),  # directory structure is preserved
}
assert moved_files.failed == {FailedRemoteFile("/source/failed.file")}
assert moved_files.skipped == {RemoteFile("/source/already.exists")}
assert moved_files.missing == {RemotePath("/source/missing.file")}

Move only certain files from source_path

from onetl.impl import RemoteFile
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"
moved_files = mover.run(
    [
        "/source/file1.txt",
        "/source/nested/path/file3.txt",
        # excluding "/source/file2.txt"
    ]
)

assert moved_files.successful == {
    RemoteFile("/target/file1.txt"),
    RemoteFile("/target/nested/path/file3.txt"),  # directory structure is preserved
}
assert not moved_files.failed
assert not moved_files.skipped
assert not moved_files.missing

Move certain files from any folder

from onetl.impl import RemoteFile
from onetl.file import FileMover

mover = FileMover(target_path="/target", ...)  # no source_path set

# only absolute paths
moved_files = mover.run(
    [
        "/remote/file1.txt",
        "/any/nested/path/file3.txt",
    ]
)

assert moved_files.successful == {
    RemoteFile("/target/file1.txt"),
    RemoteFile("/target/file3.txt"),
    # directory structure is NOT preserved without source_path
}
assert not moved_files.failed
assert not moved_files.skipped
assert not moved_files.missing

view_files() → FileSet[RemoteFile]

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

Returns:

FileSet[RemoteFile]

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

Raises:

onetl.exception.DirectoryNotFoundError

source_path does not found

NotADirectoryError

source_path is not a directory

Examples

View files

from onetl.impl import RemoteFile
from onetl.file import FileMover

mover = FileMover(source_path="/remote", ...)

view_files = mover.view_files()

assert view_files == {
    RemoteFile("/remote/file1.txt"),
    RemoteFile("/remote/file3.txt"),
    RemoteFile("/remote/nested/path/file3.txt"),
}