adamvinueza / transformer Goto Github PK

s3 = s3fs.S3FileSystem(...)
tr = Transform(src_fs=s3, dest_fs=LocalFileSystem())
tr.copy('s3://my-bucket/my-file.txt', '/path/to/my-file.txt')

from transformer.transform import Transform, S3FS, LocalFS
aws_profile='dev'
tr = Transform(src_fs=S3FS(aws_profile), dest_fs=LocalFS())
tr.copy(src, dest)

"""
class Transform(object):
    # ...

    def fcopy(self, src, dest, bufsize, filter, **kwargs):
        with self.src_fs.open(src, 'rb', bufsize) as rdr:
            with self.dest_fs.open(dest, 'wb', bufsize) as wr:
                while True:
                    b = filter(rdr, **kwargs)
                    if not b:
                        w.flush()
                        break
                    wr.write(b)
"""

# A sample filter function: encrypt an input stream.
def encrypt(src_stream, gpg, password):
    return gpg.encrypt_file(src_stream, password=passwd)

passwd = 'my-secret-gpg-password'
gpg = gnupg.GPG()
# configure gpg with key and so on


s3 = S3FS(my_profile)
tr = Transform(src_fs=s3, dest_fs=s3, overwrite=True)
tr.fcopy(src, dest, encrypt, 1024, gpg=gpg, password=passwd)

import s3fs
from fsspec.implementations import sftp

def decrypt(rdr, wr, params):
    """Decrypts the file read by rdr, writes decrypted stream using wr."""
    pass

src_fs = sftp.SFTPFileSystem(host, paramiko_params_dict)
dest_fs = s3fs.S3FileSystem(anon=False)
tr = Transform(src_fs=src_fs, dest_fs=dest_fs, overwrite=True)
tr('file_on_sftp_server.gpg', 's3://my_bucket/decrypted_file.csv', decrypt, [passphrase])

import s3fs

def decrypt(rdr, wr, params):
    """Decrypts the file read by rdr, writes decrypted stream using wr."""
    pass

fs = s3fs.S3FileSystem(anon=False)
tr = Transform(fs=fs)
tr('s3://my_bucket/encrypted.gpg', 's3://my_bucket/decrypted.csv', decrypt, [passphrase])