Source code for norfs.helpers

"""
The :mod:`norfs.helpers` offers functions that serve as shortcuts for common operations with the library.

The :mod:`norfs.helpers` contains the following functions:
"""
from typing import (
    Any,
    cast,
)
import norfs.copy.base
import norfs.copy.local
import norfs.copy.s3
import norfs.client
import norfs.fs.local
import norfs.fs.memory
import norfs.fs.s3


[docs]def get_copy_client(*args: norfs.client.FileSystemClient) -> norfs.client.CopyClient: """ Helper function to get a :class:`norfs.copy.base.CopyClient` instance configured with copy strategies for the given file systems. This function only sets the built-in :class:`norfs.copy.base.CopyStrategy` s for the built-in file systems, all other will be ignored. For example:: # Given s3_boto_1 = boto3.client('s3') s3_boto_2 = boto3.client('s3', endpoint_url='http://my.s3.endpoint') local = norfs.helpers.local() s3_1 = norfs.helpers.s3(s3_boto_1) s3_2 = norfs.helpers.s3(s3_boto_2) memory = norfs.helpers.memory() # Doing cp = get_copy_client(local, s3_1, s3_2, memory) # Is equivalent to copier = norfs.copy.base.Copier(norfs.copy.base.GenericCopyStrategy()) copier.set_copy_policy(local.fs, local.fs, norfs.copy.local.LocalToLocalCopyStrategy()) copier.set_copy_policy(local.fs, s3_1.fs, norfs.copy.local.LocalToS3CopyStrategy(s3_boto_1)) copier.set_copy_policy(local.fs, s3_2.fs, norfs.copy.local.LocalToS3CopyStrategy(s3_boto_2)) copier.set_copy_policy(s3_1.fs, local.fs, norfs.copy.s3.S3ToLocalCopyStrategy()) copier.set_copy_policy(s3_1.fs, s3_1.fs, norfs.copy.s3.S3ToS3CopyStrategy(s3_boto_1)) copier.set_copy_policy(s3_2.fs, local.fs, norfs.copy.s3.S3ToLocalCopyStrategy()) copier.set_copy_policy(s3_2.fs, s3_2.fs, norfs.copy.s3.S3ToS3CopyStrategy(s3_boto_2)) cp = norfs.client.CopyClient(copier) """ copier = norfs.copy.base.Copier(norfs.copy.base.GenericCopyStrategy()) copy_strategy: norfs.copy.base.CopyStrategy for src in args: src_type = type(src.fs) for dst in args: dst_type = type(dst.fs) if src_type is norfs.fs.local.LocalFileSystem: if dst_type is norfs.fs.local.LocalFileSystem: copy_strategy = norfs.copy.local.LocalToLocalCopyStrategy() elif dst_type is norfs.fs.s3.S3FileSystem: copy_strategy = norfs.copy.local.LocalToS3CopyStrategy( cast(norfs.fs.s3.S3FileSystem, dst.fs)._s3_client) else: continue elif src_type is norfs.fs.s3.S3FileSystem: if dst_type is norfs.fs.local.LocalFileSystem: copy_strategy = norfs.copy.s3.S3ToLocalCopyStrategy( cast(norfs.fs.s3.S3FileSystem, src.fs)._s3_client) elif dst_type is norfs.fs.s3.S3FileSystem: src_s3_client = cast(norfs.fs.s3.S3FileSystem, src.fs)._s3_client dst_s3_client = cast(norfs.fs.s3.S3FileSystem, dst.fs)._s3_client if src_s3_client is dst_s3_client: copy_strategy = norfs.copy.s3.S3ToS3CopyStrategy(src_s3_client) else: continue else: continue else: continue copier.set_copy_policy(src.fs, dst.fs, copy_strategy) return norfs.client.CopyClient(copier)
[docs]def local() -> norfs.client.FileSystemClient: """ Returns a new :class:`norfs.client.FileSystemClient` managing a new instance of :class:`norfs.fs.local.LocalFileSystem`. """ return norfs.client.FileSystemClient(norfs.fs.local.LocalFileSystem())
[docs]def memory(**kwargs: Any) -> norfs.client.FileSystemClient: """ Returns a new :class:`norfs.client.FileSystemClient` managing a new instance of :class:`norfs.fs.memory.MemoryFileSystem`. ``kwargs`` is passed directly to the file system constructor. """ return norfs.client.FileSystemClient(norfs.fs.memory.MemoryFileSystem(norfs.fs.memory.MemoryDirectory(), **kwargs))
[docs]def s3(*args: Any, **kwargs: Any) -> norfs.client.FileSystemClient: """ Returns a new :class:`norfs.client.FileSystemClient` managing a new instance of :class:`norfs.fs.s3.S3FileSystem`. ``args`` and ``kwargs`` are passed directly to the file system constructor. """ return norfs.client.FileSystemClient(norfs.fs.s3.S3FileSystem(*args, **kwargs))