Repository: Aiven-Open/rohmu (language: python)

Issue:
Fix documentation
<!-- All contributors please complete these sections, including maintainers -->
# About this change - What it does

<!-- Provide a small sentence that summarizes the change. -->

<!-- Provide the issue number below if it exists. -->
Resolves: #xxxxx

# Why this way

I noticed the `docs/`, which seems to not be generated from our `README.rst`, even though it's a copy paste of it. (duplicates stuff).
That means that even though I fixed the links [in a previous PR](https://github.com/Aiven-Open/rohmu/pull/151), this was not reflected on [our docs website](https://aiven-open.github.io/rohmu/).
There was no process for that so far and it was just copy/pasting until now.

## To do

- [ ] change copyrights from docstrings to comments
- [ ] decide on a nice API page hierarchy.


Discussion / hints:

Fix patch:
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 00000000..b1bfbfa6
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,21 @@
+*.pyo
+*.log
+*~
+/.cache
+/.coverage
+/.project
+/.pydevproject
+/.vagrant
+/.idea
+/.vscode
+__pycache__/
+/build/
+/dist/
+/rohmu.egg-info/
+/rpm/
+.vagrant
+/venv/
+/.venv/
+*.orig
+coverage.xml
+rohmu-rpm-src.tar
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 379daf63..ef7825d7 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -20,7 +20,15 @@ Contributions are very welcome on {{ PROJECT }}. When contributing please keep t
 ## Manual testing
 
 
-### Configuration
+## Building the package
+
+### On Fedora
+```shell
+sudo make fedora-dev-setup
+make rpm
+```
+
+## Configuration
 
 
 # Opening a PR
diff --git a/Makefile b/Makefile
new file mode 100644
index 00000000..0e3e2c05
--- /dev/null
+++ b/Makefile
@@ -0,0 +1,18 @@
+short_ver = $(shell git describe --abbrev=0 --always)
+long_ver = $(shell git describe --long 2>/dev/null || echo $(short_ver)-0-unknown-g`git describe --always`)
+
+.DEFAULT_GOAL := rpm
+
+.PHONY: fedora-dev-setup
+fedora-dev-setup:
+	dnf builddep -y rohmu.spec
+
+.PHONY: rpm
+rpm: rohmu/
+	git archive --output=rohmu-rpm-src.tar --prefix=rohmu/ HEAD
+	rpmbuild -bb rohmu.spec \
+		--define '_topdir $(PWD)/rpm' \
+		--define '_sourcedir $(CURDIR)' \
+		--define 'major_version $(short_ver)' \
+		--define 'minor_version $(subst -,.,$(subst $(short_ver)-,,$(long_ver)))'
+	$(RM) rohmu-rpm-src.tar
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 00000000..7e01ab27
--- /dev/null
+++ b/pyproject.toml
@@ -0,0 +1,3 @@
+[build-system]
+requires = ['setuptools', 'wheel']
+build-backend = 'setuptools.build_meta'
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 00000000..1dc64ae4
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1,1 @@
+--editable .
diff --git a/rohmu.spec b/rohmu.spec
new file mode 100644
index 00000000..d46a3d24
--- /dev/null
+++ b/rohmu.spec
@@ -0,0 +1,47 @@
+Name:           rohmu
+Version:        %{major_version}
+Release:        %{minor_version}%{?dist}
+Url:            https://github.com/aiven/rohmu
+Summary:        Object storage encryption and compression library
+License:        ASL 2.0
+Source0:        rohmu-rpm-src.tar
+Requires:       python3-botocore
+Requires:       python3-cryptography >= 0.8
+Requires:       python3-dateutil
+Requires:       python3-pydantic
+Requires:       python3-requests
+Requires:       python3-snappy
+Requires:       python3-zstandard
+BuildRequires:  python3-devel
+BuildRequires:  python3-flake8
+BuildRequires:  python3-pylint
+BuildRequires:  python3-pytest
+
+%undefine _missing_build_ids_terminate_build
+%define debug_package %{nil}
+
+%description
+Rohmu is an object storage encryption and compression library meant for backups.
+Rohmu currently supports Amazon Web Services S3, Google Cloud Storage,
+OpenStack Swift and Ceph (using S3 or Swift interfaces with RadosGW).
+Support for Microsoft Azure is experimental.
+
+%{?python_disable_dependency_generator}
+
+%prep
+%setup -q -n rohmu
+
+%build
+
+%install
+python3 setup.py install --prefix=%{_prefix} --root=%{buildroot}
+
+%files
+%defattr(-,root,root,-)
+%doc README.md
+%license LICENSE
+%{python3_sitelib}/*
+
+%changelog
+* Wed Apr 27 2022 Kevin Michel <kevin.michel@aiven.io> - 1.0.0
+- Initial RPM package spec
diff --git a/rohmu/__init__.py b/rohmu/__init__.py
new file mode 100644
index 00000000..f2a6c7b7
--- /dev/null
+++ b/rohmu/__init__.py
@@ -0,0 +1,43 @@
+"""
+rohmu
+
+Copyright (c) 2016 Ohmu Ltd
+See LICENSE for details
+"""
+from typing import Type
+
+from .errors import InvalidConfigurationError
+from .object_storage.base import BaseTransfer
+
+IO_BLOCK_SIZE = 2 ** 20  # 1 MiB
+
+
+def get_class_for_transfer(obj_store) -> Type[BaseTransfer]:
+    storage_type = obj_store["storage_type"]
+    if storage_type == "azure":
+        from .object_storage.azure import AzureTransfer
+        return AzureTransfer
+    elif storage_type == "google":
+        from .object_storage.google import GoogleTransfer
+        return GoogleTransfer
+    elif storage_type == "sftp":
+        from .object_storage.sftp import SFTPTransfer
+        return SFTPTransfer
+    elif storage_type == "local":
+        from .object_storage.local import LocalTransfer
+        return LocalTransfer
+    elif storage_type == "s3":
+        from .object_storage.s3 import S3Transfer
+        return S3Transfer
+    elif storage_type == "swift":
+        from .object_storage.swift import SwiftTransfer
+        return SwiftTransfer
+
+    raise InvalidConfigurationError("unsupported storage type {0!r}".format(storage_type))
+
+
+def get_transfer(storage_config) -> BaseTransfer:
+    storage_class = get_class_for_transfer(storage_config)
+    storage_config = storage_config.copy()
+    storage_config.pop("storage_type")
+    return storage_class(**storage_config)
diff --git a/rohmu/compat.py b/rohmu/compat.py
new file mode 100644
index 00000000..b6de081f
--- /dev/null
+++ b/rohmu/compat.py
@@ -0,0 +1,17 @@
+# rohmu - This module used to contain compatible implementations for older
+# python version.
+#
+# Since we don't support those older versions anymore, this module could be removed,
+# but as rohmu is used outside pghoard itself better keep the imports and throw
+# a deprecation warning.
+
+import warnings
+from contextlib import suppress  # pylint: disable=unused-import
+from os import makedirs  # pylint: disable=unused-import
+
+warnings.warn(
+    "pghoard.rohmu.compat is deprecated, you should import "
+    "from the standard library directly instead",
+    DeprecationWarning,
+    stacklevel=2
+)
diff --git a/rohmu/compressor.py b/rohmu/compressor.py
new file mode 100644
index 00000000..08d20a38
--- /dev/null
+++ b/rohmu/compressor.py
@@ -0,0 +1,100 @@
+"""
+rohmu - compressor interface
+
+Copyright (c) 2016 Ohmu Ltd
+See LICENSE for details
+"""
+
+import lzma
+
+from .errors import InvalidConfigurationError
+from .filewrap import Sink, Stream
+from .snappyfile import SnappyFile
+from .zstdfile import open as zstd_open
+
+try:
+    import snappy
+except ImportError:
+    snappy = None  # type: ignore
+
+try:
+    import zstandard as zstd
+except ImportError:
+    zstd = None  # type: ignore
+
+
+def CompressionFile(dst_fp, algorithm, level=0, threads=0):
+    """This looks like a class to users, but is actually a function that instantiates a class based on algorithm."""
+    if algorithm == "lzma":
+        return lzma.open(dst_fp, "w", preset=level)
+
+    if algorithm == "snappy":
+        return SnappyFile(dst_fp, "wb")
+
+    if algorithm == "zstd":
+        return zstd_open(dst_fp, "wb", level=level, threads=threads)
+
+    if algorithm:
+        raise InvalidConfigurationError("invalid compression algorithm: {!r}".format(algorithm))
+
+    return dst_fp
+
+
+class CompressionStream(Stream):
+    """Non-seekable stream of data that adds compression on top of given source stream"""
+    def __init__(self, src_fp, algorithm, level=0):
+        super().__init__(src_fp, minimum_read_size=32 * 1024)
+        if algorithm == "lzma":
+            self._compressor = lzma.LZMACompressor(lzma.FORMAT_XZ, -1, level, None)
+        elif algorithm == "snappy":
+            self._compressor = snappy.StreamCompressor()
+        elif algorithm == "zstd":
+            self._compressor = zstd.ZstdCompressor(level=level).compressobj()
+        else:
+            raise InvalidConfigurationError("invalid compression algorithm: {!r}".format(algorithm))
+
+    def _process_chunk(self, data):
+        return self._compressor.compress(data)
+
+    def _finalize(self):
+        return self._compressor.flush()
+
+
+def DecompressionFile(src_fp, algorithm):
+    """This looks like a class to users, but is actually a function that instantiates a class based on algorithm."""
+    if algorithm == "lzma":
+        return lzma.open(src_fp, "r")
+
+    if algorithm == "snappy":
+        return SnappyFile(src_fp, "rb")
+
+    if algorithm == "zstd":
+        return zstd_open(src_fp, "rb")
+
+    if algorithm:
+        raise InvalidConfigurationError("invalid compression algorithm: {!r}".format(algorithm))
+
+    return src_fp
+
+
+class DecompressSink(Sink):
+    def __init__(self, next_sink, compression_algorithm):
+        super().__init__(next_sink)
+        self.decompressor = self._create_decompressor(compression_algorithm)
+
+    def _create_decompressor(self, alg):
+        if alg == "snappy":
+            return snappy.StreamDecompressor()
+        elif alg == "lzma":
+            return lzma.LZMADecompressor()
+        elif alg == "zstd":
+            return zstd.ZstdDecompressor().decompressobj()
+        raise InvalidConfigurationError("invalid compression algorithm: {!r}".format(alg))
+
+    def write(self, data):
+        written = len(data)
+        if not data:
+            return written
+        data = self.decompressor.decompress(data)
+        self._write_to_next_sink(data)
+        return written
diff --git a/rohmu/dates.py b/rohmu/dates.py
new file mode 100644
index 00000000..f85e1a22
--- /dev/null
+++ b/rohmu/dates.py
@@ -0,0 +1,45 @@
+"""
+rohmu - timestamp handling
+
+Copyright (c) 2017 Ohmu Ltd
+See LICENSE for details
+"""
+import datetime
+
+import dateutil.parser
+import dateutil.tz
+
+
+def parse_timestamp(ts, *, with_tz=True, assume_local=False):
+    """Parse a given timestamp and return a datetime object with or without tzinfo.
+
+    If `with_tz` is False and we can't parse a timezone from the timestamp the datetime object is returned
+    as-is and we assume the timezone is whatever was requested.  If `with_tz` is False and we can parse a
+    timezone, the timestamp is converted to either local or UTC time based on `assume_local` after which tzinfo
+    is stripped and the timestamp is returned.
+
+    When `with_tz` is True and there's a timezone in the timestamp we return it as-is.  If `with_tz` is True
+    but we can't parse a timezone we add either local or UTC timezone to the datetime based on `assume_local`.
+    """
+    parse_result = dateutil.parser.parse(ts)
+
+    # pylint thinks dateutil.parser.parse always returns a tuple even though we didn't request it.
+    # So this check is pointless but convinces pylint that we really have a datetime object now.
+    dt = parse_result[0] if isinstance(parse_result, tuple) else parse_result  # pylint: disable=unsubscriptable-object
+
+    if with_tz is False:
+        if not dt.tzinfo:
+            return dt
+
+        tz = dateutil.tz.tzlocal() if assume_local else datetime.timezone.utc
+        return dt.astimezone(tz).replace(tzinfo=None)
+
+    if dt.tzinfo:
+        return dt
+
+    tz = dateutil.tz.tzlocal() if assume_local else datetime.timezone.utc
+    return dt.replace(tzinfo=tz)
+
+
+def now():
+    return datetime.datetime.now(datetime.timezone.utc)
diff --git a/rohmu/delta/__init__.py b/rohmu/delta/__init__.py
new file mode 100644
index 00000000..956777b8
--- /dev/null
+++ b/rohmu/delta/__init__.py
@@ -0,0 +1,1 @@
+# Copyright (c) 2021 Aiven, Helsinki, Finland. https://aiven.io/
diff --git a/rohmu/delta/common.py b/rohmu/delta/common.py
new file mode 100644
index 00000000..adecdf8b
--- /dev/null
+++ b/rohmu/delta/common.py
@@ -0,0 +1,282 @@
+# Copyright (c) 2021 Aiven, Helsinki, Finland. https://aiven.io/
+import functools
+import hashlib
+import json as _json
+import logging
+import math
+import os
+from datetime import datetime
+from multiprocessing.dummy import Pool
+from pathlib import Path
+from typing import List, Optional
+
+from pydantic import BaseModel, Field
+
+from rohmu.dates import now
+
+_hash = hashlib.blake2s
+_log_1_1 = math.log(1.1)
+
+# Hexdigest is 32 bytes, so something orders of magnitude more at least
+EMBEDDED_FILE_SIZE = 150
+
+logger = logging.getLogger(__name__)
+
+
+def hash_hexdigest_readable(f, *, read_buffer=1_000_000):
+    h = _hash()
+    while True:
+        data = f.read(read_buffer)
+        if not data:
+            break
+        h.update(data)
+    return h.hexdigest()
+
+
+def increase_worth_reporting(value, new_value=None, *, total=None):
+    """ Make reporting sparser and sparser as values grow larger
+    - report every 1.1**N or so
+    - if we know total, report every percent
+    """
+    if new_value is None:
+        new_value = value
+        value = new_value - 1
+    if total is not None:
+        if new_value == total or total <= 100:
+            return True
+        old_percent = 100 * value // total
+        new_percent = 100 * new_value // total
+        return old_percent != new_percent
+    if value <= 10 or new_value <= 10:
+        return True
+    old_exp = int(math.log(value) / _log_1_1)
+    new_exp = int(math.log(new_value) / _log_1_1)
+    return old_exp != new_exp
+
+
+class DeltaModel(BaseModel):
+    class Config:
+        # As we're keen to both export and decode json, just using
+        # enum values for encode/decode is much saner than the default
+        # enumname.value (it is also slightly less safe but oh well)
+        use_enum_values = True
+
+        # Extra values should be errors, as they are most likely typos
+        # which lead to grief when not detected. However, if we ever
+        # start deprecating some old fields and not wanting to parse
+        # them, this might need to be revisited.
+        extra = "forbid"
+
+        # Validate field default values too
+        validate_all = True
+
+        # Validate also assignments
+        # validate_assignment = True
+        # TBD: Figure out why this doesn't work in some unit tests;
+        # possibly the tests themselves are broken
+
+    def jsondict(self, **kw):
+        # By default,
+        #
+        # .json() returns json string.
+        # .dict() returns Python dict, but it has things that are not
+        # json serializable.
+        #
+        # We provide json seralizable dict (super inefficiently) here.
+        #
+        # This is mostly used for test code so that should be fine
+        return _json.loads(self.json(**kw))
+
+
+class SizeLimitedFile:
+    def __init__(self, *, path, file_size):
+        self._f = open(path, "rb")
+        self._file_size = file_size
+        self.tell = self._f.tell
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, t, v, tb):
+        self._f.close()
+
+    def read(self, n=None):
+        can_read = max(0, self._file_size - self._f.tell())
+        if n is None:
+            n = can_read
+        n = min(can_read, n)
+        return self._f.read(n)
+
+    def seek(self, ofs, whence=0):
+        if whence == os.SEEK_END:
+            ofs += self._file_size
+            whence = os.SEEK_SET
+        return self._f.seek(ofs, whence)
+
+
+class SnapshotHash(DeltaModel):
+    """
+    This class represents something that is to be stored in the object storage.
+
+    size is provided mainly to allow for even loading of nodes in case
+    same hexdigest is available from multiple nodes.
+
+    """
+    hexdigest: str
+    size: int
+
+    def __eq__(self, other):
+        if isinstance(other, SnapshotHash):
+            return self.hexdigest == other.hexdigest
+        return False
+
+    def __hash__(self):
+        # hexdigests should be unique, regardless of size
+        return hash(self.hexdigest)
+
+
+@functools.total_ordering
+class SnapshotFile(DeltaModel):
+    relative_path: Path
+    file_size: int
+    stored_file_size: int
+    mtime_ns: int
+    hexdigest: str = ""
+    content_b64: Optional[str]
+
+    def __lt__(self, o):
+        # In our use case, paths uniquely identify files we care about
+        return self.relative_path < o.relative_path
+
+    def equals_excluding_mtime(self, o):
+        return self.copy(update={"mtime_ns": 0}) == o.copy(update={"mtime_ns": 0})
+
+    def open_for_reading(self, root_path):
+        return SizeLimitedFile(path=root_path / self.relative_path, file_size=self.file_size)
+
+
+class SnapshotState(DeltaModel):
+    root_globs: List[str]
+    files: List[SnapshotFile]
+    empty_dirs: List[Path]
+
+
+class SnapshotResult(DeltaModel):
+    # when was the operation started ( / done )
+    start: datetime = Field(default_factory=now)
+    end: Optional[datetime]
+    #
+    # should be passed opaquely to restore
+    state: Optional[SnapshotState]
+    #
+    # Summary data for manifest use
+    files: int = 0
+    total_size: int = 0
+
+    # populated only if state is available
+    hashes: Optional[List[SnapshotHash]]
+
+
+class SnapshotUploadResult(DeltaModel):
+    total_size: int = 0
+    total_stored_size: int = 0
+
+
+class BackupManifest(DeltaModel):
+    start: datetime
+    end: datetime = Field(default_factory=now)
+
+    # Filesystem snapshot contents of the backup
+    snapshot_result: SnapshotResult
+
+    # What did the upload return (mostly for statistics)
+    upload_result: SnapshotUploadResult
+
+
+class Progress(DeltaModel):
+    """ JSON-encodable progress meter of sorts """
+    handled: int = 0
+    failed: int = 0
+    total: int = 0
+    final: bool = False
+
+    def __repr__(self):
+        finished = ", finished" if self.final else ""
+        return f"{self.handled}/{self.total} handled, {self.failed} failures{finished}"
+
+    def start(self, n):
+        " Optional 'first' step, just for logic handling state (e.g. no progress object reuse desired) "
+        assert not self.total
+        logger.debug("start")
+        self.add_total(n)
+
+    def add_total(self, n):
+        if not n:
+            return
+        old_total = self.total
+        self.total += n
+        if increase_worth_reporting(old_total, self.total):
+            logger.debug("add_total %r -> %r", n, self)
+        assert not self.final
+
+    def add_fail(self, n=1, *, info="add_fail"):
+        assert n > 0
+        old_failed = self.failed
+        self.failed += n
+        if increase_worth_reporting(old_failed, self.failed):
+            logger.debug("%s %r -> %r", info, n, self)
+        assert not self.final
+
+    def add_success(self, n=1, *, info="add_success"):
+        assert n > 0
+        old_handled = self.handled
+        self.handled += n
+        assert self.handled <= self.total
+        if increase_worth_reporting(old_handled, self.handled, total=self.total):
+            logger.debug("%s %r -> %r", info, n, self)
+        assert not self.final
+
+    def download_success(self, size):
+        self.add_success(size, info="download_success")
+
+    def upload_success(self, hexdigest):
+        self.add_success(info=f"upload_success {hexdigest}")
+
+    def upload_missing(self, hexdigest):
+        self.add_fail(info=f"upload_missing {hexdigest}")
+
+    def upload_failure(self, hexdigest):
+        self.add_fail(info=f"upload_failure {hexdigest}")
+
+    def done(self):
+        assert self.total is not None and self.handled <= self.total
+        assert not self.final
+        self.final = True
+        logger.debug("done %r", self)
+
+    @property
+    def finished_successfully(self):
+        return self.final and not self.failed and self.handled == self.total
+
+    @property
+    def finished_failed(self):
+        return self.final and not self.finished_successfully
+
+    @classmethod
+    def merge(cls, progresses):
+        p = cls()
+        for progress in progresses:
+            p.handled += progress.handled
+            p.failed += progress.failed
+            p.total += progress.total
+        p.final = all(progress.final for progress in progresses)
+        return p
+
+
+def parallel_map_to(*, fun, iterable, result_callback, n=None) -> bool:
+    iterable_as_list = list(iterable)
+    with Pool(n) as p:
+        for map_in, map_out in zip(iterable_as_list, p.imap(fun, iterable_as_list)):
+            if not result_callback(map_in=map_in, map_out=map_out):
+                return False
+    return True
diff --git a/rohmu/delta/snapshot.py b/rohmu/delta/snapshot.py
new file mode 100644
index 00000000..84c38874
--- /dev/null
+++ b/rohmu/delta/snapshot.py
@@ -0,0 +1,238 @@
+# Copyright (c) 2021 Aiven, Helsinki, Finland. https://aiven.io/
+import base64
+import logging
+import os
+import threading
+from pathlib import Path
+from typing import Callable, Dict, List, Optional
+
+from rohmu.delta.common import (
+    EMBEDDED_FILE_SIZE, Progress, SnapshotFile, SnapshotHash, SnapshotState, hash_hexdigest_readable,
+    increase_worth_reporting, parallel_map_to
+)
+
+logger = logging.getLogger(__name__)
+
+
+class Snapshotter:
+    """Snapshotter keeps track of files on disk, and their hashes.
+
+        The hash on disk MAY change, which may require subsequent
+        incremential snapshot and-or ignoring the files which have changed.
+
+        The output to outside is just root object's hash, as well as list
+        of other hashes which correspond to files referred to within the
+        file list contained in root object.
+
+        Note that any call to public API MUST be made with
+        snapshotter.lock held. This is because Snapshotter is process-wide
+        utility that is shared across operations, possibly used from
+        multiple threads, and the single-operation-only mode of operation
+        is not exactly flawless (the 'new operation can be started with
+        old running' is intentional feature but new operation should
+        eventually replace the old). The lock itself might not need to be
+        built-in to Snapshotter, but having it there enables asserting its
+        state during public API calls.
+        """
+    def __init__(self, *, src, dst, globs, src_iterate_func: Optional[Callable] = None, parallel=1):
+        assert globs
+        self.src = Path(src)
+        self.dst = Path(dst)
+        self.globs = globs
+        self.src_iterate_func = src_iterate_func
+        self.relative_path_to_snapshotfile: Dict[Path, SnapshotFile] = {}
+        self.hexdigest_to_snapshotfiles: Dict[str, List[SnapshotFile]] = {}
+        self.parallel = parallel
+        self.lock = threading.Lock()
+        self.empty_dirs: List[Path] = []
+
+    def _list_files(self, basepath: Path):
+        result_files = set()
+        for glob in self.globs:
+            for path in basepath.glob(glob):
+                if not path.is_file() or path.is_symlink():
+                    continue
+                relpath = path.relative_to(basepath)
+                result_files.add(relpath)
+
+        return sorted(result_files)
+
+    def _list_dirs_and_files(self, basepath: Path):
+        files = self._list_files(basepath)
+        dirs = {p.parent for p in files}
+        return sorted(dirs), files
+
+    def _add_snapshotfile(self, snapshotfile: SnapshotFile):
+        old_snapshotfile = self.relative_path_to_snapshotfile.get(snapshotfile.relative_path, None)
+        if old_snapshotfile:
+            self._remove_snapshotfile(old_snapshotfile)
+        self.relative_path_to_snapshotfile[snapshotfile.relative_path] = snapshotfile
+        if snapshotfile.hexdigest:
+            self.hexdigest_to_snapshotfiles.setdefault(snapshotfile.hexdigest, []).append(snapshotfile)
+
+    def _remove_snapshotfile(self, snapshotfile: SnapshotFile):
+        assert self.relative_path_to_snapshotfile[snapshotfile.relative_path] == snapshotfile
+        del self.relative_path_to_snapshotfile[snapshotfile.relative_path]
+        if snapshotfile.hexdigest:
+            self.hexdigest_to_snapshotfiles[snapshotfile.hexdigest].remove(snapshotfile)
+
+    def _snapshotfile_from_path(self, relative_path):
+        src_path = self.src / relative_path
+        st = src_path.stat()
+        return SnapshotFile(relative_path=relative_path, mtime_ns=st.st_mtime_ns, file_size=st.st_size, stored_file_size=0)
+
+    def _gen_snapshot_hashes(self, relative_paths, reuse_old_snapshotfiles):
+        same = 0
+        lost = 0
+        for relative_path in relative_paths:
+            old_snapshotfile = self.relative_path_to_snapshotfile.get(relative_path)
+            try:
+                snapshotfile = self._snapshotfile_from_path(relative_path)
+            except FileNotFoundError:
+                lost += 1
+                if increase_worth_reporting(lost):
+                    logger.debug("#%d. lost - %s disappeared before stat, ignoring", lost, self.src / relative_path)
+                continue
+            if reuse_old_snapshotfiles and old_snapshotfile:
+                snapshotfile.hexdigest = old_snapshotfile.hexdigest
+                snapshotfile.content_b64 = old_snapshotfile.content_b64
+                if old_snapshotfile == snapshotfile:
+                    same += 1
+                    if increase_worth_reporting(same):
+                        logger.debug("#%d. same - %r in %s is same", same, old_snapshotfile, relative_path)
+                    continue
+            yield snapshotfile
+
+    def get_snapshot_hashes(self):
+        assert self.lock.locked()
+        return [
+            SnapshotHash(hexdigest=dig, size=sf[0].file_size) for dig, sf in self.hexdigest_to_snapshotfiles.items() if sf
+        ]
+
+    def get_snapshot_state(self):
+        assert self.lock.locked()
+        return SnapshotState(
+            root_globs=self.globs, files=sorted(self.relative_path_to_snapshotfile.values()), empty_dirs=self.empty_dirs
+        )
+
+    def update_snapshot_file_data(self, *, relative_path, hexdigest, file_size, stored_file_size):
+        snapshotfile = self.relative_path_to_snapshotfile[relative_path]
+        snapshotfile.hexdigest = hexdigest
+        snapshotfile.file_size = file_size
+        snapshotfile.stored_file_size = stored_file_size
+
+    def _snapshot_create_missing_directories(self, *, src_dirs, dst_dirs):
+        changes = 0
+        for i, relative_dir in enumerate(set(src_dirs).difference(dst_dirs), 1):
+            dst_path = self.dst / relative_dir
+            dst_path.mkdir(parents=True, exist_ok=True)
+            if increase_worth_reporting(i):
+                logger.debug("#%d. new directory: %r", i, relative_dir)
+            changes += 1
+        return changes
+
+    def _snapshot_remove_extra_files(self, *, src_files, dst_files):
+        changes = 0
+        for i, relative_path in enumerate(set(dst_files).difference(src_files), 1):
+            dst_path = self.dst / relative_path
+            snapshotfile = self.relative_path_to_snapshotfile.get(relative_path)
+            if snapshotfile:
+                self._remove_snapshotfile(snapshotfile)
+            dst_path.unlink()
+            if increase_worth_reporting(i):
+                logger.debug("#%d. extra file: %r", i, relative_path)
+            changes += 1
+        return changes
+
+    def _snapshot_add_missing_files(self, *, src_files, dst_files):
+        existing = 0
+        disappeared = 0
+        changes = 0
+        for i, relative_path in enumerate(set(src_files).difference(dst_files), 1):
+            src_path = self.src / relative_path
+            dst_path = self.dst / relative_path
+            try:
+                os.link(src=src_path, dst=dst_path, follow_symlinks=False)
+            except FileExistsError:
+                # This happens only if snapshot is started twice at
+                # same time. While it is technically speaking upstream
+                # error, we rather handle it here than leave
+                # exceptions not handled.
+                existing += 1
+                if increase_worth_reporting(existing):
+                    logger.debug("#%d. %s already existed, ignoring", existing, src_path)
+                continue
+            except FileNotFoundError:
+                disappeared += 1
+                if increase_worth_reporting(disappeared):
+                    logger.debug("#%d. %s disappeared before linking, ignoring", disappeared, src_path)
+                continue
+            if increase_worth_reporting(i - disappeared):
+                logger.debug("#%d. new file: %r", i - disappeared, relative_path)
+            changes += 1
+        return changes
+
+    def snapshot(self, *, progress: Optional[Progress] = None, reuse_old_snapshotfiles=True):
+        assert self.lock.locked()
+
+        if progress is None:
+            progress = Progress()
+        progress.start(3)
+
+        if self.src_iterate_func:
+            src_dirs_set = set()
+            src_files_set = set()
+            for item in self.src_iterate_func():
+                path = Path(item)
+                if path.is_file() and not path.is_symlink():
+                    src_files_set.add(path.relative_to(self.src))
+                elif path.is_dir():
+                    src_dirs_set.add(path.relative_to(self.src))
+
+            src_dirs = sorted(src_dirs_set | {p.parent for p in src_files_set})
+            src_files = sorted(src_files_set)
+        else:
+            src_dirs, src_files = self._list_dirs_and_files(self.src)
+
+        dst_dirs, dst_files = self._list_dirs_and_files(self.dst)
+
+        # Create missing directories
+        changes = self._snapshot_create_missing_directories(src_dirs=src_dirs, dst_dirs=dst_dirs)
+        progress.add_success()
+
+        # Remove extra files
+        changes += self._snapshot_remove_extra_files(src_files=src_files, dst_files=dst_files)
+        progress.add_success()
+
+        # Add missing files
+        changes += self._snapshot_add_missing_files(src_files=src_files, dst_files=dst_files)
+        progress.add_success()
+
+        # We COULD also remove extra directories, but it is not
+        # probably really worth it and due to ignored files it
+        # actually might not even work.
+
+        # Then, create/update corresponding snapshotfile objects (old
+        # ones were already removed)
+        dst_dirs, dst_files = self._list_dirs_and_files(self.dst)
+        self.empty_dirs = src_dirs
+        snapshotfiles = list(self._gen_snapshot_hashes(dst_files, reuse_old_snapshotfiles))
+        progress.add_total(len(snapshotfiles))
+
+        def _cb(snapshotfile):
+            # src may or may not be present; dst is present as it is in snapshot
+            with snapshotfile.open_for_reading(self.dst) as f:
+                if snapshotfile.file_size <= EMBEDDED_FILE_SIZE:
+                    snapshotfile.content_b64 = base64.b64encode(f.read()).decode()
+                else:
+                    snapshotfile.hexdigest = hash_hexdigest_readable(f)
+            return snapshotfile
+
+        def _result_cb(*, map_in, map_out):
+            self._add_snapshotfile(map_out)
+            progress.add_success()
+            return True
+
+        changes += len(snapshotfiles)
+        parallel_map_to(iterable=snapshotfiles, fun=_cb, result_callback=_result_cb, n=self.parallel)
+        return changes
diff --git a/rohmu/encryptor.py b/rohmu/encryptor.py
new file mode 100644
index 00000000..d03c7920
--- /dev/null
+++ b/rohmu/encryptor.py
@@ -0,0 +1,408 @@
+"""
+rohmu - content encryption
+
+Copyright (c) 2016 Ohmu Ltd
+See LICENSE for details
+"""
+
+import io
+import logging
+import os
+import struct
+
+import cryptography
+import cryptography.hazmat.backends.openssl.backend
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import padding
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+from cryptography.hazmat.primitives.hashes import SHA1, SHA256
+from cryptography.hazmat.primitives.hmac import HMAC
+
+from . import IO_BLOCK_SIZE
+from .filewrap import FileWrap, Sink, Stream
+
+if cryptography.__version__ < "1.6":  # type: ignore
+    # workaround for deadlock https://github.com/pyca/cryptography/issues/2911
+    cryptography.hazmat.backends.openssl.backend.activate_builtin_random()
+
+FILEMAGIC = b"pghoa1"
+AES_BLOCK_SIZE = 16
+
+
+class EncryptorError(Exception):
+    """ EncryptorError """
+
+
+class Encryptor:
+    def __init__(self, rsa_public_key_pem):
+        if not isinstance(rsa_public_key_pem, bytes):
+            rsa_public_key_pem = rsa_public_key_pem.encode("ascii")
+        self.rsa_public_key = serialization.load_pem_public_key(rsa_public_key_pem, backend=default_backend())
+        self.cipher = None
+        self.authenticator = None
+
+    def update(self, data):
+        ret = b""
+        if self.cipher is None:
+            key = os.urandom(16)
+            nonce = os.urandom(16)
+            auth_key = os.urandom(32)
+            self.cipher = Cipher(algorithms.AES(key), modes.CTR(nonce), backend=default_backend()).encryptor()
+            self.authenticator = HMAC(auth_key, SHA256(), backend=default_backend())
+            pad = padding.OAEP(mgf=padding.MGF1(algorithm=SHA1()), algorithm=SHA1(), label=None)
+            cipherkey = self.rsa_public_key.encrypt(key + nonce + auth_key, pad)
+            ret = FILEMAGIC + struct.pack(">H", len(cipherkey)) + cipherkey
+        cur = self.cipher.update(data)
+        self.authenticator.update(cur)
+        if ret:
+            return ret + cur
+        else:
+            return cur
+
+    def finalize(self):
+        if self.cipher is None:
+            return b""  # empty plaintext input yields empty encrypted output
+
+        ret = self.cipher.finalize()
+        self.authenticator.update(ret)
+        ret += self.authenticator.finalize()
+        self.cipher = None
+        self.authenticator = None
+        return ret
+
+
+class EncryptorFile(FileWrap):
+    def __init__(self, next_fp, rsa_public_key_pem):
+        super().__init__(next_fp)
+        self.key = rsa_public_key_pem
+        self.encryptor = Encryptor(self.key)
+        self.offset = 0
+        self.state = "OPEN"
+
+    def flush(self):
+        self._check_not_closed()
+        self.next_fp.flush()
+
+    def close(self):
+        if self.state == "CLOSED":
+            return
+        final = self.encryptor.finalize()
+        self.encryptor = None
+        self.next_fp.write(final)
+        super().close()
+
+    def writable(self):
+        """True if this stream supports writing"""
+        self._check_not_closed()
+        return True
+
+    def write(self, data):
+        """Encrypt and write the given bytes"""
+        self._check_not_closed()
+        if not data:
+            return 0
+        enc_data = self.encryptor.update(data)
+        self.next_fp.write(enc_data)
+        self.offset += len(data)
+        return len(data)
+
+
+class EncryptorStream(Stream):
+    """Non-seekable stream of data that adds encryption on top of given source stream"""
+    def __init__(self, src_fp, rsa_public_key_pem):
+        super().__init__(src_fp)
+        self._encryptor = Encryptor(rsa_public_key_pem)
+
+    def _process_chunk(self, data):
+        return self._encryptor.update(data)
+
+    def _finalize(self):
+        return self._encryptor.finalize()
+
+
+class Decryptor:
+    def __init__(self, rsa_private_key_pem):
+        if not isinstance(rsa_private_key_pem, bytes):
+            rsa_private_key_pem = rsa_private_key_pem.encode("ascii")
+        self.rsa_private_key = serialization.load_pem_private_key(
+            data=rsa_private_key_pem, password=None, backend=default_backend()
+        )
+        self.cipher = None
+        self.authenticator = None
+        self._cipher_key_len = None
+        self._header_size = None
+        self._footer_size = 32
+
+    def expected_header_bytes(self):
+        if self._header_size is not None:
+            return 0
+        return self._cipher_key_len or 8
+
+    def header_size(self):
+        return self._header_size
+
+    def footer_size(self):
+        return self._footer_size
+
+    def process_header(self, data):
+        if self._cipher_key_len is None:
+            if data[0:6] != FILEMAGIC:
+                raise EncryptorError("Invalid magic bytes")
+            self._cipher_key_len = struct.unpack(">H", data[6:8])[0]
+        else:
+            pad = padding.OAEP(mgf=padding.MGF1(algorithm=SHA1()), algorithm=SHA1(), label=None)
+            try:
+                plainkey = self.rsa_private_key.decrypt(data, pad)
+            except AssertionError:
+                raise EncryptorError("Decrypting key data failed")
+            if len(plainkey) != 64:
+                raise EncryptorError("Integrity check failed")
+            key = plainkey[0:16]
+            nonce = plainkey[16:32]
+            auth_key = plainkey[32:64]
+            self._header_size = 8 + len(data)
+
+            self.cipher = Cipher(algorithms.AES(key), modes.CTR(nonce), backend=default_backend()).decryptor()
+            self.authenticator = HMAC(auth_key, SHA256(), backend=default_backend())
+
+    def process_data(self, data):
+        if not data:
+            return b""
+        self.authenticator.update(data)
+        return self.cipher.update(data)
+
+    def finalize(self, footer):
+        if footer != self.authenticator.finalize():
+            raise EncryptorError("Integrity check failed")
+        result = self.cipher.finalize()
+        self.cipher = None
+        self.authenticator = None
+        return result
+
+
+class DecryptorFile(FileWrap):
+    def __init__(self, next_fp, rsa_private_key_pem):
+        super().__init__(next_fp)
+        self._key = rsa_private_key_pem
+        self.log = logging.getLogger(self.__class__.__name__)
+        self._decryptor = None
+        self._crypted_size = None
+        self._boundary_block = None
+        self._plaintext_size = None
+        # Our actual plain-text read offset. seek may change self.offset to something
+        # else temporarily but we keep _decrypt_offset intact until we actually do a
+        # read in case the caller just called seek in order to then immediately seek back
+        self._decrypt_offset = None
+        self.offset = None
+        self._reset()
+
+    def _reset(self):
+        self._decryptor = Decryptor(self._key)
+        self._crypted_size = self._file_size(self.next_fp)
+        self._boundary_block = None
+        self._plaintext_size = None
+        self._decrypt_offset = 0
+        # Plaintext offset
+        self.offset = 0
+        self.state = "OPEN"
+
+    @classmethod
+    def _file_size(cls, file):
+        current_offset = file.seek(0, os.SEEK_SET)
+        file_end_offset = file.seek(0, os.SEEK_END)
+        file.seek(current_offset, os.SEEK_SET)
+        return file_end_offset
+
+    def _initialize_decryptor(self):
+        if self._plaintext_size is not None:
+            return
+        while True:
+            required_bytes = self._decryptor.expected_header_bytes()
+            if not required_bytes:
+                break
+            self._decryptor.process_header(self._read_raw_exactly(required_bytes))
+        self._plaintext_size = self._crypted_size - self._decryptor.header_size() - self._decryptor.footer_size()
+
+    def _read_raw_exactly(self, required_bytes):
+        data = self.next_fp.read(required_bytes)
+        while data and len(data) < required_bytes:
+            next_chunk = self.next_fp.read(required_bytes - len(data))
+            if not next_chunk:
+                break
+            data += next_chunk
+        if not data or len(data) != required_bytes:
+            raise EncryptorError("Failed to read {} bytes of header or footer data".format(required_bytes))
+        return data
+
+    def _move_decrypt_offset_to_plaintext_offset(self):
+        if self._decrypt_offset == self.offset:
+            return
+        seek_to = self.offset
+        if self._decrypt_offset > self.offset:
+            self.log.warning("Negative seek from %d to %d, must re-initialize decryptor", self._decrypt_offset, self.offset)
+            self._reset()
+            self._initialize_decryptor()
+        discard_bytes = seek_to - self._decrypt_offset
+        self.offset = self._decrypt_offset
+        while discard_bytes > 0:
+            data = self._read_block(discard_bytes)
+            discard_bytes -= len(data)
+
+    def _read_all(self):
+        full_data = bytearray()
+        while True:
+            data = self._read_block(IO_BLOCK_SIZE)
+            if not data:
+                return bytes(full_data)
+            full_data.extend(data)
+
+    def _read_block(self, size):
+        if self._crypted_size == 0:
+            return b""
+
+        self._initialize_decryptor()
+
+        if self.offset == self._plaintext_size:
+            return b""
+
+        self._move_decrypt_offset_to_plaintext_offset()
+
+        # If we have an existing boundary block, fulfil the read entirely from that
+        if self._boundary_block:
+            size = min(size, len(self._boundary_block) - self.offset % AES_BLOCK_SIZE)
+            data = self._boundary_block[self.offset % AES_BLOCK_SIZE:self.offset % AES_BLOCK_SIZE + size]
+            if self.offset % AES_BLOCK_SIZE + size == len(self._boundary_block):
+                self._boundary_block = None
+            data_len = len(data)
+            self.offset += data_len
+            self._decrypt_offset += data_len
+            return data
+
+        # Only serve multiples of AES_BLOCK_SIZE whenever possible to keep things simpler
+        read_size = size
+        if self.offset + max(AES_BLOCK_SIZE, size) >= self._plaintext_size:
+            read_size = self._plaintext_size - self.offset
+        elif size > AES_BLOCK_SIZE and size % AES_BLOCK_SIZE != 0 and self.offset + size < self._plaintext_size:
+            read_size = size - size % AES_BLOCK_SIZE
+        elif size < AES_BLOCK_SIZE:
+            read_size = AES_BLOCK_SIZE
+
+        encrypted = self._read_raw_exactly(read_size)
+        decrypted = self._decryptor.process_data(encrypted)
+        if self.offset + read_size == self._plaintext_size:
+            footer = self._read_raw_exactly(self._decryptor.footer_size())
+            last_part = self._decryptor.finalize(footer)
+            if last_part:
+                decrypted += last_part
+
+        if size < AES_BLOCK_SIZE:
+            self._boundary_block = decrypted
+            return self._read_block(size)
+        decrypted_len = len(decrypted)
+        self.offset += decrypted_len
+        self._decrypt_offset += decrypted_len
+        return decrypted
+
+    def close(self):
+        super().close()
+        self._decryptor = None
+
+    def read(self, size=-1):
+        """Read up to size decrypted bytes"""
+        self._check_not_closed()
+        if self.state == "EOF" or size == 0:
+            return b""
+        elif size < 0:
+            return self._read_all()
+        else:
+            return self._read_block(size)
+
+    def readable(self):
+        """True if this stream supports reading"""
+        self._check_not_closed()
+        return self.state in ["OPEN", "EOF"]
+
+    def seek(self, offset, whence=0):
+        self._check_not_closed()
+        self._initialize_decryptor()
+        if whence == os.SEEK_SET:
+            if offset != self.offset:
+                if offset > self._plaintext_size:
+                    raise io.UnsupportedOperation("DecryptorFile does not support seeking beyond EOF")
+                if offset < 0:
+                    raise ValueError("negative seek position")
+                self.offset = offset
+            return self.offset
+        elif whence == os.SEEK_CUR:
+            if offset != 0:
+                raise io.UnsupportedOperation("can't do nonzero cur-relative seeks")
+            return self.offset
+        elif whence == os.SEEK_END:
+            if offset != 0:
+                raise io.UnsupportedOperation("can't do nonzero end-relative seeks")
+            self.offset = self._plaintext_size
+            return self.offset
+        else:
+            raise ValueError("Invalid whence value")
+
+    def seekable(self):
+        """True if this stream supports random access"""
+        self._check_not_closed()
+        return True
+
+
+class DecryptSink(Sink):
+    def __init__(self, next_sink, file_size, encryption_key_data):
+        super().__init__(next_sink)
+        if file_size <= 0:
+            raise ValueError("Invalid file_size: " + str(file_size))
+        self.data_bytes_received = 0
+        self.data_size = file_size
+        self.decryptor = Decryptor(encryption_key_data)
+        self.file_size = file_size
+        self.footer = b""
+        self.header = b""
+
+    def _extract_encryption_footer_bytes(self, data):
+        expected_data_bytes = self.data_size - self.data_bytes_received
+        if len(data) > expected_data_bytes:
+            self.footer += data[expected_data_bytes:]
+            data = data[:expected_data_bytes]
+        return data
+
+    def _process_encryption_header(self, data):
+        if not data or not self.decryptor.expected_header_bytes():
+            return data
+        if self.header:
+            data = self.header + data
+            self.header = None
+        offset = 0
+        while self.decryptor.expected_header_bytes() > 0:
+            header_bytes = self.decryptor.expected_header_bytes()
+            if header_bytes + offset > len(data):
+                self.header = data[offset:]
+                return b""
+            self.decryptor.process_header(data[offset:offset + header_bytes])
+            offset += header_bytes
+        data = data[offset:]
+        self.data_size = self.file_size - self.decryptor.header_size() - self.decryptor.footer_size()
+        return data
+
+    def write(self, data):
+        written = len(data)
+        data = self._process_encryption_header(data)
+        if not data:
+            return written
+        data = self._extract_encryption_footer_bytes(data)
+        self.data_bytes_received += len(data)
+        if data:
+            data = self.decryptor.process_data(data)
+        if len(self.footer) == self.decryptor.footer_size():
+            final_data = self.decryptor.finalize(self.footer)
+            if final_data:
+                data += final_data
+        if not data:
+            return written
+        self._write_to_next_sink(data)
+        return written
diff --git a/rohmu/errors.py b/rohmu/errors.py
new file mode 100644
index 00000000..3c1bcb40
--- /dev/null
+++ b/rohmu/errors.py
@@ -0,0 +1,34 @@
+"""
+rohmu - exception classes
+
+Copyright (c) 2016 Ohmu Ltd
+See LICENSE for details
+"""
+
+
+class Error(Exception):
+    """Generic exception"""
+
+
+class StorageError(Error):
+    """Storage exception"""
+
+
+class FileNotFoundFromStorageError(StorageError):
+    """File not found from remote storage"""
+
+
+class InvalidConfigurationError(Error):
+    """Invalid configuration"""
+
+
+class LocalFileIsRemoteFileError(StorageError):
+    """File transfer operation source and destination point to the same file"""
+
+
+class MissingLibraryError(Exception):
+    """Missing dependency library"""
+
+
+class MaybeRecoverableError(Error):
+    """An error that may be recoverable"""
diff --git a/rohmu/filewrap.py b/rohmu/filewrap.py
new file mode 100644
index 00000000..8f5aa0eb
--- /dev/null
+++ b/rohmu/filewrap.py
@@ -0,0 +1,189 @@
+"""
+rohmu - file transformation wrapper
+
+Copyright (c) 2016 Ohmu Ltd
+See LICENSE for details
+"""
+
+import io
+import time
+
+
+class FileWrap(io.BufferedIOBase):
+    # pylint: disable=unused-argument
+
+    def __init__(self, next_fp):
+        super().__init__()
+        self.next_fp = next_fp
+        self.offset = 0
+        self.state = "OPEN"
+
+    def _check_not_closed(self):
+        if self.state == "CLOSED":
+            raise ValueError("I/O operation on closed file")
+
+    def close(self):
+        """Close stream"""
+        if self.state == "CLOSED":
+            return
+        self.flush()
+        # We close the stack of rohmu file wrappers, but leave the underlying real io object open to allow the
+        # caller to do something useful with it if they like, for example reading the output out of a BytesIO
+        # object or linking a temporary file to another name, etc.
+        if isinstance(self.next_fp, FileWrap):
+            self.next_fp.close()
+        self.next_fp = None
+        self.state = "CLOSED"
+
+    @property
+    def closed(self):
+        """True if this stream is closed"""
+        return self.state == "CLOSED"
+
+    def fileno(self):
+        self._check_not_closed()
+        return self.next_fp.fileno()
+
+    def flush(self):
+        self._check_not_closed()
+
+    def tell(self):
+        self._check_not_closed()
+        return self.offset
+
+    def readable(self):
+        """True if this stream supports reading"""
+        self._check_not_closed()
+        return False
+
+    def read(self, size=-1):
+        """Read up to size decrypted bytes"""
+        self._check_not_closed()
+        raise io.UnsupportedOperation("Read not supported")
+
+    def seekable(self):
+        """True if this stream supports random access"""
+        self._check_not_closed()
+        return False
+
+    def seek(self, offset, whence=0):
+        self._check_not_closed()
+        raise io.UnsupportedOperation("Seek not supported")
+
+    def truncate(self):
+        self._check_not_closed()
+        raise io.UnsupportedOperation("Truncate not supported")
+
+    def writable(self):
+        """True if this stream supports writing"""
+        self._check_not_closed()
+        return False
+
+    def write(self, data):
+        """Encrypt and write the given bytes"""
+        self._check_not_closed()
+        raise io.UnsupportedOperation("Write not supported")
+
+
+class Sink:
+    """Sink performs transformation for received input data and passes it forward to
+    given target sink. Data is fed to this class via it's `write` method and that in
+    turn calls `write` for the next sink.
+
+    Unlike the FileWrap interface, which is useful for performing transformation when
+    data is being pulled from a source, the Sink interface can be used when data is
+    pushed through a pipeline of transformations. This provides better performance as
+    any temporary files or buffers can be omitted."""
+    def __init__(self, next_sink):
+        self.next_sink = next_sink
+
+    def _data_written(self, bytes_written, pending_bytes):
+        pass
+
+    def _write_to_next_sink(self, data):
+        data = memoryview(data)
+        offset = 0
+        while offset < len(data):
+            start_offset = offset
+            offset += self.next_sink.write(data[offset:])
+            self._data_written(offset - start_offset, len(data) - offset)
+
+    def write(self, data):
+        """Performs some transformation for given data and writes the transformed
+        data to next sink."""
+        self._write_to_next_sink(data)
+        return len(data)
+
+
+class ThrottleSink(Sink):
+    """Provides simple throttling sink that can be used if the target sink is
+    non-blocking device that can return short if all data cannot be immediately
+    written. In such cases writing again immediately after a small write would
+    result in unnecessary busy-looping."""
+    def __init__(self, next_sink, wait_time, sleep_fn=time.sleep):
+        super().__init__(next_sink)
+        self.sleep_fn = sleep_fn
+        self.wait_time = wait_time
+
+    def _data_written(self, bytes_written, pending_bytes):
+        if pending_bytes > 0 and bytes_written < 10 * 1024:
+            self.sleep_fn(self.wait_time)
+
+
+class Stream:
+    """Non-seekable stream of data that performs some kind of processing for given source stream"""
+    def __init__(self, src_fp, *, minimum_read_size=8 * 1024):
+        self._eof = False
+        self._remainder = b""
+        self._src = src_fp
+        self.minimum_read_size = minimum_read_size
+        self._offset = 0
+
+    def _process_chunk(self, data):
+        raise NotImplementedError
+
+    def _finalize(self):
+        raise NotImplementedError
+
+    def read(self, size=-1):
+        bytes_available = len(self._remainder)
+        chunks = [self._remainder] if self._remainder else []
+        while not self._eof and (size < 0 or bytes_available < size):
+            bytes_to_read = -1 if size < 0 else size - bytes_available
+            # Always read at least self.minimum_read_size bytes even if fewer bytes are requested to avoid
+            # looping with very small buffers when stream processor needs non-trivial amount of input to
+            # make progress
+            if 0 < bytes_to_read < self.minimum_read_size:
+                bytes_to_read = self.minimum_read_size
+            src_data = self._src.read(bytes_to_read)
+            if not src_data:
+                dst_data = self._finalize()
+            else:
+                dst_data = self._process_chunk(src_data)
+            if dst_data:
+                chunks.append(dst_data)
+                bytes_available += len(dst_data)
+            if not src_data:
+                self._eof = True
+
+        if size < 0 or bytes_available < size:
+            data = b"".join(chunks)
+            self._remainder = b""
+        else:
+            # We only read up to one chunk beyond the required amount of data so all but
+            # the last chunk will necessarily always be included in the response
+            data = b"".join(chunks[:-1])
+            bytes_missing = size - len(data)
+            if bytes_missing == len(chunks[-1]):
+                data += chunks[-1]
+                self._remainder = b""
+            elif bytes_missing > 0:
+                data += chunks[-1][:bytes_missing]
+                self._remainder = chunks[-1][bytes_missing:]
+            else:
+                self._remainder = chunks[-1]
+        self._offset += len(data)
+        return data
+
+    def tell(self):
+        return self._offset
diff --git a/rohmu/inotify.py b/rohmu/inotify.py
new file mode 100644
index 00000000..9c428019
--- /dev/null
+++ b/rohmu/inotify.py
@@ -0,0 +1,153 @@
+"""
+rohmu - inotify wrapper
+
+Copyright (c) 2016 Ohmu Ltd
+See LICENSE for details
+"""
+import ctypes
+import logging
+import os
+import select
+import struct
+from contextlib import suppress
+from ctypes import c_char_p, c_int, c_uint32
+from threading import Thread
+
+
+class InotifyEvent(ctypes.Structure):
+    _fields_ = [
+        ("wd", c_int),
+        ("mask", c_uint32),
+        ("cookie", c_uint32),
+        ("len", c_uint32),
+        ("name", c_char_p),
+    ]
+
+
+s_size = 16
+# default 2048 events
+INOTIFY_EVENT_BUFFER_SIZE = 2048 * (ctypes.sizeof(InotifyEvent) + s_size)
+
+event_types = {
+    "IN_CLOSE_WRITE": 0x00000008,
+    "IN_CREATE": 0x00000100,
+    "IN_MOVED_FROM": 0x00000040,
+    "IN_MOVED_TO": 0x00000080,
+    "IN_DELETE": 0x00000200,
+    "IN_DELETE_SELF": 0x00000400,
+    "IN_IGNORED": 0x8000
+}
+
+IN_NONBLOCK = 0x00004000
+
+
+def parse_inotify_buffer(event_buffer):
+    i = 0
+    while i + s_size <= len(event_buffer):
+        wd, mask, cookie, length = struct.unpack_from("iIII", event_buffer, i)
+        name = event_buffer[i + s_size:i + s_size + length].rstrip(b"\0")
+        i += s_size + length
+        yield wd, mask, cookie, name
+
+
+class InotifyWatcher(Thread):
+    def __init__(self, compression_queue):
+        super().__init__()
+        # use the newer form for future-proofness
+        self.log = logging.getLogger("PGHoardInotify")
+        self.libc = ctypes.CDLL("libc.so.6", use_errno=True)
+        self.fd = self.libc.inotify_init()
+        self.watch_to_path = {}
+        self.cookies = {}
+        self.running = True
+        self.compression_queue = compression_queue
+        self.timeout = 1.0
+        self.log.debug("InotifyWatcher initialized")
+
+    def add_watch(self, path, events=None):
+        mask = 0
+        events = events or event_types.keys()
+        for key in events:
+            mask |= event_types[key]
+        watch = self.libc.inotify_add_watch(self.fd, c_char_p(path.encode("utf8")), c_uint32(mask))
+        if watch < 0:
+            errno = ctypes.get_errno()
+            raise OSError(errno, os.strerror(errno))
+        self.watch_to_path[watch] = path
+        self.log.debug("Added watch for path: %r", path)
+
+    def read_events(self):
+        event_buffer = None
+        while self.running:
+            with suppress(InterruptedError):
+                rlist, _, _ = select.select([self.fd], [], [], self.timeout)
+                if rlist:
+                    for fd in rlist:
+                        event_buffer = os.read(fd, INOTIFY_EVENT_BUFFER_SIZE)
+                break
+        if not event_buffer:
+            return
+        for wd, mask, cookie, name in parse_inotify_buffer(event_buffer):
+            if wd == -1:
+                continue
+            self.create_event(wd, mask, cookie, name)
+
+    def log_event(self, ev_type, full_path):
+        if self.log.getEffectiveLevel() > logging.DEBUG:
+            return
+
+        try:
+            st = os.stat(full_path)
+        except:  # pylint: disable=bare-except
+            st = None
+
+        self.log.debug("event: %s %s, %r", full_path, ev_type, st)
+
+    def create_event(self, wd, mask, cookie, name):
+        if mask & event_types["IN_IGNORED"]:
+            # explicit removal of watch or dir, ignore
+            return
+
+        decoded_name = name.decode("utf8")
+        watched_path = self.watch_to_path[wd]
+        full_path = os.path.join(self.watch_to_path[wd], decoded_name)
+
+        if mask & event_types["IN_CREATE"] > 0:
+            # file was created but writing to it is not finished yet
+            self.log_event("IN_CREATE", full_path)
+        elif mask & event_types["IN_CLOSE_WRITE"] > 0:
+            # file was open for writing and was closed
+            self.log_event("IN_CLOSE_WRITE", full_path)
+            self.compression_queue.put({"type": "CLOSE_WRITE", "full_path": full_path, "watched_path": watched_path})
+        elif mask & event_types["IN_DELETE"] > 0:
+            self.log_event("IN_DELETE", full_path)
+            self.compression_queue.put({"type": "DELETE", "full_path": full_path, "watched_path": watched_path})
+
+        elif mask & event_types["IN_DELETE_SELF"] > 0:
+            # the monitored directory was deleted
+            self.log_event("IN_DELETE_SELF", full_path)
+            directory = self.watch_to_path.pop(wd, "")
+            self.log.debug("Directory: %r that we were watching has been deleted, removing watch", directory)
+            self.libc.inotify_rm_watch(self.fd, wd)
+        elif mask & event_types["IN_MOVED_FROM"] > 0:
+            self.log_event("IN_MOVED_FROM", full_path)
+            self.cookies[cookie] = full_path
+        elif mask & event_types["IN_MOVED_TO"] > 0:
+            self.log_event("IN_MOVED_TO", full_path)
+            src_path = self.cookies.pop(cookie, None)
+            if src_path:
+                self.compression_queue.put({
+                    "type": "MOVE",
+                    "full_path": full_path,
+                    "src_path": src_path,
+                    "watched_path": watched_path
+                })
+            else:
+                self.compression_queue.put({"type": "CREATE", "full_path": full_path, "watched_path": watched_path})
+
+    def run(self):
+        self.log.debug("Starting InotifyWatcher")
+        while self.running:
+            self.read_events()
+        self.log.debug("Quitting InotifyWatcher")
+        os.close(self.fd)
diff --git a/rohmu/object_storage/__init__.py b/rohmu/object_storage/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/rohmu/object_storage/azure.py b/rohmu/object_storage/azure.py
new file mode 100644
index 00000000..54364c6a
--- /dev/null
+++ b/rohmu/object_storage/azure.py
@@ -0,0 +1,357 @@
+"""
+rohmu - azure object store interface
+
+Copyright (c) 2016 Ohmu Ltd
+See LICENSE for details
+"""
+import logging
+import time
+from io import BytesIO
+
+# pylint: disable=import-error, no-name-in-module
+import azure.common
+from azure.core.exceptions import ResourceExistsError
+from azure.storage.blob import BlobServiceClient, ContentSettings
+
+try:
+    from azure.storage.blob import BlobPrefix, BlobType
+except ImportError:
+    # old versions of the azure blob storage library do not expose the classes publicly
+    from azure.storage.blob._models import BlobPrefix, BlobType
+
+from ..errors import (FileNotFoundFromStorageError, InvalidConfigurationError, StorageError)
+from .base import (KEY_TYPE_OBJECT, KEY_TYPE_PREFIX, BaseTransfer, IterKeyItem, get_total_memory)
+
+ENDPOINT_SUFFIXES = {
+    None: "core.windows.net",
+    "germany": "core.cloudapi.de",  # Azure Germany is a completely separate cloud from the regular Azure Public cloud
+    "china": "core.chinacloudapi.cn",
+    "public": "core.windows.net"
+}
+
+
+def calculate_max_block_size():
+    total_mem_mib = get_total_memory() or 0
+    # At least 4 MiB, at most 100 MiB. Max block size used for hosts with ~100+ GB of memory
+    return max(min(int(total_mem_mib / 1000), 100), 4) * 1024 * 1024
+
+
+# Increase block size based on host memory. Azure supports up to 50k blocks and up to 5 TiB individual
+# files. Default block size is set to 4 MiB so only ~200 GB files can be uploaded. In order to get close
+# to that 5 TiB increase the block size based on host memory; we don't want to use the max 100 for all
+# hosts because the uploader will allocate (with default settings) 3 x block size of memory.
+MAX_BLOCK_SIZE = calculate_max_block_size()
+
+# Reduce Azure logging verbocity of http requests and responses
+logging.getLogger("azure.core.pipeline.policies.http_logging_policy").setLevel(logging.WARNING)
+
+
+class AzureTransfer(BaseTransfer):
+    def __init__(self, account_name, account_key, bucket_name, prefix=None, azure_cloud=None, proxy_info=None):
+        prefix = "{}".format(prefix.lstrip("/") if prefix else "")
+        super().__init__(prefix=prefix)
+        self.account_name = account_name
+        self.account_key = account_key
+        self.container_name = bucket_name
+        try:
+            endpoint_suffix = ENDPOINT_SUFFIXES[azure_cloud]
+        except KeyError:
+            raise InvalidConfigurationError("Unknown azure cloud {!r}".format(azure_cloud))
+
+        conn_str = (
+            "DefaultEndpointsProtocol=https;"
+            f"AccountName={self.account_name};"
+            f"AccountKey={self.account_key};"
+            f"EndpointSuffix={endpoint_suffix}"
+        )
+        config = {"max_block_size": MAX_BLOCK_SIZE}
+        if proxy_info:
+            username = proxy_info.get("user")
+            password = proxy_info.get("pass")
+            if username and password:
+                auth = f"{username}:{password}@"
+            else:
+                auth = ""
+            host = proxy_info["host"]
+            port = proxy_info["port"]
+            if proxy_info.get("type") == "socks5":
+                schema = "socks5"
+            else:
+                schema = "http"
+            config["proxies"] = {"https": f"{schema}://{auth}{host}:{port}"}
+
+        self.conn = BlobServiceClient.from_connection_string(
+            conn_str=conn_str,
+            **config,
+        )
+        self.conn.socket_timeout = 120  # Default Azure socket timeout 20s is a bit short
+        self.container = self.get_or_create_container(self.container_name)
+        self.log.debug("AzureTransfer initialized, %r", self.container_name)
+
+    def copy_file(self, *, source_key, destination_key, metadata=None, **kwargs):
+        timeout = kwargs.get("timeout") or 15
+        source_path = self.format_key_for_backend(source_key, remove_slash_prefix=True, trailing_slash=False)
+        destination_path = self.format_key_for_backend(destination_key, remove_slash_prefix=True, trailing_slash=False)
+        source_client = self.conn.get_blob_client(self.container_name, source_path)
+        destination_client = self.conn.get_blob_client(self.container_name, destination_path)
+        source_url = source_client.url
+        start = time.monotonic()
+
+        destination_client.start_copy_from_url(source_url, metadata=metadata, timeout=timeout)
+        while True:
+            blob_properties = destination_client.get_blob_properties(timeout=timeout)
+            copy_props = blob_properties.copy
+            if copy_props.status == "success":
+                return
+            elif copy_props.status == "pending":
+                if time.monotonic() - start < timeout:
+                    time.sleep(0.1)
+                else:
+                    destination_client.abort_copy(copy_props.id, timeout=timeout)
+                    raise StorageError(
+                        "Copying {!r} to {!r} did not complete in {} seconds".format(source_key, destination_key, timeout)
+                    )
+            elif copy_props.status == "failed":
+                raise StorageError(
+                    "Copying {!r} to {!r} failed: {!r}".format(source_key, destination_key, copy_props.status_description)
+                )
+            else:
+                raise StorageError(
+                    "Copying {!r} to {!r} failed, unexpected status: {!r}".format(
+                        source_key, destination_key, copy_props.status
+                    )
+                )
+
+    def get_metadata_for_key(self, key):
+        path = self.format_key_for_backend(key, remove_slash_prefix=True, trailing_slash=False)
+        items = list(self._iter_key(path=path, with_metadata=True, deep=False))
+        if not items:
+            raise FileNotFoundFromStorageError(key)
+        expected_name = path.rsplit("/", 1)[-1]
+        for item in items:
+            # We expect single result but Azure listing is prefix match so we need to explicitly
+            # look up the matching result
+            if item.type == KEY_TYPE_OBJECT:
+                item_name = item.value["name"]
+            else:
+                item_name = item.value
+            if item_name.rstrip("/").rsplit("/", 1)[-1] == expected_name:
+                break
+        else:
+            item = None
+        if not item or item.type != KEY_TYPE_OBJECT:
+            raise FileNotFoundFromStorageError(key)  # not found or prefix
+        return item.value["metadata"]
+
+    def _metadata_for_key(self, path):
+        return list(self._iter_key(path=path, with_metadata=True, deep=False))[0].value["metadata"]
+
+    def iter_key(self, key, *, with_metadata=True, deep=False, include_key=False):
+        path = self.format_key_for_backend(key, remove_slash_prefix=True, trailing_slash=not include_key)
+        self.log.debug("Listing path %r", path)
+        yield from self._iter_key(path=path, with_metadata=with_metadata, deep=deep)
+
+    def _iter_key(self, *, path, with_metadata, deep):
+        include = "metadata" if with_metadata else None
+        container_client = self.conn.get_container_client(self.container_name)
+        name_starts_with = None
+        delimiter = ""
+        if path:
+            # If you give Azure an empty path, it gives you an authentication error
+            name_starts_with = path
+        if not deep:
+            delimiter = "/"
+        items = container_client.walk_blobs(include=include, name_starts_with=name_starts_with, delimiter=delimiter)
+        for item in items:
+            if isinstance(item, BlobPrefix):
+                yield IterKeyItem(type=KEY_TYPE_PREFIX, value=self.format_key_from_backend(item.name).rstrip("/"))
+            else:
+                if with_metadata:
+                    metadata = {}
+                    if item.metadata:
+                        # Azure Storage cannot handle '-' so we turn them into underscores and back again
+                        metadata = {k.replace("_", "-"): v for k, v in item.metadata.items()}
+                else:
+                    metadata = None
+                yield IterKeyItem(
+                    type=KEY_TYPE_OBJECT,
+                    value={
+                        "last_modified": item.last_modified,
+                        "metadata": metadata,
+                        "name": self.format_key_from_backend(item.name),
+                        "size": item.size,
+                    },
+                )
+
+    def delete_key(self, key):
+        key = self.format_key_for_backend(key, remove_slash_prefix=True)
+        self.log.debug("Deleting key: %r", key)
+        try:
+            blob_client = self.conn.get_blob_client(container=self.container_name, blob=key)
+            return blob_client.delete_blob()
+        except azure.core.exceptions.ResourceNotFoundError as ex:  # pylint: disable=no-member
+            raise FileNotFoundFromStorageError(key) from ex
+
+    def get_contents_to_file(self, key, filepath_to_store_to, *, progress_callback=None):
+        key = self.format_key_for_backend(key, remove_slash_prefix=True)
+
+        self.log.debug("Starting to fetch the contents of: %r to: %r", key, filepath_to_store_to)
+        try:
+            with open(filepath_to_store_to, "wb") as f:
+                container_client = self.conn.get_container_client(self.container_name)
+                container_client.download_blob(key).readinto(f)
+        except azure.core.exceptions.ResourceNotFoundError as ex:  # pylint: disable=no-member
+            raise FileNotFoundFromStorageError(key) from ex
+
+        if progress_callback:
+            progress_callback(1, 1)
+        return self._metadata_for_key(key)
+
+    @classmethod
+    def _parse_length_from_content_range(cls, content_range):
+        """Parses the blob length from the content range header: bytes 1-3/65537"""
+        if not content_range:
+            raise ValueError("File size unavailable")
+
+        return int(content_range.split(" ", 1)[1].split("/", 1)[1])
+
+    def _stream_blob(self, key, fileobj, progress_callback):
+        """Streams contents of given key to given fileobj. Data is read sequentially in chunks
+        without any seeks. This requires duplicating some functionality of the Azure SDK, which only
+        allows reading entire blob into memory at once or returning data from random offsets"""
+        file_size = None
+        start_range = 0
+        chunk_size = self.conn._config.max_chunk_get_size  # pylint: disable=protected-access
+        end_range = chunk_size - 1
+        blob = self.conn.get_blob_client(self.container_name, key)
+        while True:
+            try:
+                # pylint: disable=protected-access
+                download_stream = blob.download_blob(offset=start_range, length=chunk_size)
+                if file_size is None:
+                    file_size = download_stream._file_size
+                download_stream.readinto(fileobj)
+                start_range += download_stream.size
+                if start_range >= file_size:
+                    break
+                if download_stream.size == 0:
+                    raise StorageError("Empty response received for {}, range {}-{}".format(key, start_range, end_range))
+                end_range += download_stream.size
+                if end_range >= file_size:
+                    end_range = file_size - 1
+                if progress_callback:
+                    progress_callback(start_range, file_size)
+            except azure.core.exceptions.ResourceNotFoundError as ex:  # pylint: disable=no-member
+                if ex.status_code == 416:  # Empty file
+                    return
+                raise FileNotFoundFromStorageError(key) from ex
+
+    def get_contents_to_fileobj(self, key, fileobj_to_store_to, *, progress_callback=None):
+        key = self.format_key_for_backend(key, remove_slash_prefix=True)
+
+        self.log.debug("Starting to fetch the contents of: %r", key)
+        try:
+            self._stream_blob(key, fileobj_to_store_to, progress_callback)
+        except azure.core.exceptions.ResourceNotFoundError as ex:  # pylint: disable=no-member
+            raise FileNotFoundFromStorageError(key) from ex
+
+        if progress_callback:
+            progress_callback(1, 1)
+        return self._metadata_for_key(key)
+
+    def get_contents_to_string(self, key):
+        key = self.format_key_for_backend(key, remove_slash_prefix=True)
+        self.log.debug("Starting to fetch the contents of: %r", key)
+        try:
+            container_client = self.conn.get_container_client(self.container_name)
+            blob = BytesIO()
+            container_client.download_blob(key).download_to_stream(blob)
+            return blob.getvalue(), self._metadata_for_key(key)
+        except azure.core.exceptions.ResourceNotFoundError as ex:  # pylint: disable=no-member
+            raise FileNotFoundFromStorageError(key) from ex
+
+    def get_file_size(self, key):
+        key = self.format_key_for_backend(key, remove_slash_prefix=True)
+        try:
+            blob_client = self.conn.get_blob_client(self.container_name, key)
+            return blob_client.get_blob_properties().size
+        except azure.core.exceptions.ResourceNotFoundError as ex:  # pylint: disable=no-member
+            raise FileNotFoundFromStorageError(key) from ex
+
+    def store_file_from_memory(self, key, memstring, metadata=None, cache_control=None, mimetype=None):
+        if cache_control is not None:
+            raise NotImplementedError("AzureTransfer: cache_control support not implemented")
+        key = self.format_key_for_backend(key, remove_slash_prefix=True)
+        content_settings = None
+        if mimetype:
+            content_settings = ContentSettings(content_type=mimetype)
+        blob_client = self.conn.get_blob_client(self.container_name, key)
+        blob_client.upload_blob(
+            bytes(memstring),
+            blob_type=BlobType.BlockBlob,
+            content_settings=content_settings,
+            metadata=self.sanitize_metadata(metadata, replace_hyphen_with="_"),
+            overwrite=True
+        )
+
+    def store_file_from_disk(self, key, filepath, metadata=None, multipart=None, cache_control=None, mimetype=None):
+        if cache_control is not None:
+            raise NotImplementedError("AzureTransfer: cache_control support not implemented")
+        key = self.format_key_for_backend(key, remove_slash_prefix=True)
+        content_settings = None
+        if mimetype:
+            content_settings = ContentSettings(content_type=mimetype)
+        with open(filepath, "rb") as data:
+            blob_client = self.conn.get_blob_client(self.container_name, key)
+            blob_client.upload_blob(
+                data,
+                blob_type=BlobType.BlockBlob,
+                content_settings=content_settings,
+                metadata=self.sanitize_metadata(metadata, replace_hyphen_with="_"),
+                overwrite=True
+            )
+
+    def store_file_object(self, key, fd, *, cache_control=None, metadata=None, mimetype=None, upload_progress_fn=None):
+        if cache_control is not None:
+            raise NotImplementedError("AzureTransfer: cache_control support not implemented")
+        key = self.format_key_for_backend(key, remove_slash_prefix=True)
+        content_settings = None
+        if mimetype:
+            content_settings = ContentSettings(content_type=mimetype)
+
+        def progress_callback(pipeline_response):
+            bytes_sent = pipeline_response.context["upload_stream_current"]
+            if upload_progress_fn and bytes_sent:
+                upload_progress_fn(bytes_sent)
+
+        # Azure _BlobChunkUploader calls `tell()` on the stream even though it doesn't use the result.
+        # We expect the input stream not to support `tell()` so use dummy implementation for it
+        seekable = hasattr(fd, "seekable") and fd.seekable()
+        if not seekable:
+            original_tell = getattr(fd, "tell", None)
+            fd.tell = lambda: None
+        try:
+            blob_client = self.conn.get_blob_client(self.container_name, key)
+            blob_client.upload_blob(
+                fd,
+                blob_type=BlobType.BlockBlob,
+                content_settings=content_settings,
+                metadata=self.sanitize_metadata(metadata, replace_hyphen_with="_"),
+                raw_response_hook=progress_callback,
+                overwrite=True
+            )
+        finally:
+            if not seekable:
+                if original_tell is not None:
+                    fd.tell = original_tell
+                else:
+                    delattr(fd, "tell")
+
+    def get_or_create_container(self, container_name):
+        start_time = time.monotonic()
+        try:
+            self.conn.create_container(container_name)
+        except ResourceExistsError:
+            pass
+        self.log.debug("Got/Created container: %r successfully, took: %.3fs", container_name, time.monotonic() - start_time)
+        return container_name
diff --git a/rohmu/object_storage/base.py b/rohmu/object_storage/base.py
new file mode 100644
index 00000000..53cd8202
--- /dev/null
+++ b/rohmu/object_storage/base.py
@@ -0,0 +1,137 @@
+"""
+rohmu - object_storage.base
+
+Copyright (c) 2016 Ohmu Ltd
+See LICENSE for details
+"""
+import logging
+import platform
+from collections import namedtuple
+
+from ..errors import StorageError
+
+KEY_TYPE_OBJECT = "object"
+KEY_TYPE_PREFIX = "prefix"
+
+IterKeyItem = namedtuple("IterKeyItem", ["type", "value"])
+
+
+class BaseTransfer:
+    def __init__(self, prefix):
+        self.log = logging.getLogger(self.__class__.__name__)
+        if not prefix:
+            prefix = ""
+        elif prefix[-1] != "/":
+            prefix += "/"
+        self.prefix = prefix
+
+    def copy_file(self, *, source_key, destination_key, metadata=None, **_kwargs):
+        """Performs remote copy from source key name to destination key name. Key must identify a file, trees
+        cannot be copied with this method. If no metadata is given copies the existing metadata."""
+        raise NotImplementedError
+
+    def format_key_for_backend(self, key, remove_slash_prefix=False, trailing_slash=False):
+        """Add a possible prefix to the key before sending it to the backend"""
+        path = self.prefix + key
+        if trailing_slash:
+            if not path or path[-1] != "/":
+                path += "/"
+        else:
+            path = path.rstrip("/")
+        if remove_slash_prefix:  # Azure defines slashes in the beginning as "dirs" for listing purposes
+            path = path.lstrip("/")
+        return path
+
+    def format_key_from_backend(self, key):
+        """Strip the configured prefix from a key retrieved from the backend
+        before passing it on to other pghoard code and presenting it to the
+        user."""
+        if not self.prefix:
+            return key
+        if not key.startswith(self.prefix):
+            raise StorageError("Key {!r} does not start with expected prefix {!r}".format(key, self.prefix))
+        return key[len(self.prefix):]
+
+    def delete_key(self, key):
+        raise NotImplementedError
+
+    def delete_tree(self, key):
+        """Delete all keys under given root key. Basic implementation works by just listing all available
+        keys and deleting them individually but storage providers can implement more efficient logic."""
+        self.log.debug("Deleting tree: %r", key)
+        names = [item["name"] for item in self.list_path(key, with_metadata=False, deep=True)]
+        for name in names:
+            self.delete_key(name)
+
+    def get_contents_to_file(self, key, filepath_to_store_to, *, progress_callback=None):
+        """Write key contents to file pointed by `path` and return metadata.  If `progress_callback` is
+        provided it must be a function which accepts two numeric arguments: current state of progress and the
+        expected maximum value.  The actual values and value ranges differ per storage provider, some (S3)
+        reporting the number of bytes transmitted as the first argument and the total number of expected bytes
+        as the second argument, while others (Google) report the progress in percentage as the first value and
+        100 as the second value."""
+        raise NotImplementedError
+
+    def get_contents_to_fileobj(self, key, fileobj_to_store_to, *, progress_callback=None):
+        """Like `get_contents_to_file()` but writes to an open file-like object."""
+        raise NotImplementedError
+
+    def get_contents_to_string(self, key):
+        """Returns a tuple (content-byte-string, metadata)"""
+        raise NotImplementedError
+
+    def get_file_size(self, key):
+        """Returns an int indicating the size of the file in bytes"""
+        # This method isn't currently used by PGHoard itself, it is merely provided
+        # for applications that use PGHoard's object storage abstraction layer.
+        raise NotImplementedError
+
+    def get_metadata_for_key(self, key):
+        raise NotImplementedError
+
+    def list_path(self, key, *, with_metadata=True, deep=False):
+        return list(self.list_iter(key, with_metadata=with_metadata, deep=deep))
+
+    def list_iter(self, key, *, with_metadata=True, deep=False):
+        for item in self.iter_key(key, with_metadata=with_metadata, deep=deep):
+            if item.type == KEY_TYPE_OBJECT:
+                yield item.value
+
+    def list_prefixes(self, key):
+        return list(self.iter_prefixes(key))
+
+    def iter_prefixes(self, key):
+        for item in self.iter_key(key, with_metadata=False):
+            if item.type == KEY_TYPE_PREFIX:
+                yield item.value
+
+    def iter_key(self, key, *, with_metadata=True, deep=False, include_key=False):
+        raise NotImplementedError
+
+    def sanitize_metadata(self, metadata, replace_hyphen_with="-"):
+        """Convert non-string metadata values to strings and drop null values"""
+        return {str(k).replace("-", replace_hyphen_with): str(v) for k, v in (metadata or {}).items() if v is not None}
+
+    def store_file_from_memory(self, key, memstring, metadata=None, cache_control=None, mimetype=None):
+        raise NotImplementedError
+
+    def store_file_from_disk(self, key, filepath, metadata=None, multipart=None, cache_control=None, mimetype=None):
+        raise NotImplementedError
+
+    def store_file_object(self, key, fd, *, cache_control=None, metadata=None, mimetype=None, upload_progress_fn=None):
+        raise NotImplementedError
+
+
+def get_total_memory():
+    """return total system memory in mebibytes (or None if parsing meminfo fails)"""
+    if platform.system() != "Linux":
+        return None
+
+    with open("/proc/meminfo", "r") as in_file:
+        for line in in_file:
+            info = line.split()
+            if info[0] == "MemTotal:" and info[-1] == "kB":
+                memory_mb = int(int(info[1]) / 1024)
+                return memory_mb
+
+    return None
diff --git a/rohmu/object_storage/google.py b/rohmu/object_storage/google.py
new file mode 100644
index 00000000..7eaed0e3
--- /dev/null
+++ b/rohmu/object_storage/google.py
@@ -0,0 +1,538 @@
+"""
+rohmu - google cloud object store interface
+
+Copyright (c) 2016 Ohmu Ltd
+See LICENSE for details
+"""
+# pylint: disable=import-error, no-name-in-module
+
+import codecs
+import errno
+import json
+import logging
+import os
+import random
+import socket
+import ssl
+import time
+from contextlib import contextmanager
+from http.client import IncompleteRead
+from io import BytesIO, FileIO
+
+# NOTE: this import is not needed per-se, but it's imported here first to point the
+# user to the most important possible missing dependency
+import googleapiclient  # noqa pylint: disable=unused-import
+import httplib2
+from googleapiclient.discovery import build
+from googleapiclient.errors import HttpError
+from googleapiclient.http import (MediaFileUpload, MediaIoBaseDownload, MediaIoBaseUpload, MediaUpload, build_http)
+from oauth2client import GOOGLE_TOKEN_URI
+from oauth2client.client import GoogleCredentials
+
+from ..dates import parse_timestamp
+from ..errors import FileNotFoundFromStorageError, InvalidConfigurationError
+from .base import (KEY_TYPE_OBJECT, KEY_TYPE_PREFIX, BaseTransfer, IterKeyItem, get_total_memory)
+
+try:
+    from oauth2client.service_account import ServiceAccountCredentials
+    ServiceAccountCredentials_from_dict = ServiceAccountCredentials.from_json_keyfile_dict
+except ImportError:
+    from oauth2client.service_account import _ServiceAccountCredentials
+
+    def ServiceAccountCredentials_from_dict(credentials, scopes=None):
+        if scopes is None:
+            scopes = []
+        return _ServiceAccountCredentials(
+            service_account_id=credentials["client_id"],
+            service_account_email=credentials["client_email"],
+            private_key_id=credentials["private_key_id"],
+            private_key_pkcs8_text=credentials["private_key"],
+            scopes=scopes
+        )
+
+
+# Silence Google API client verbose spamming
+logging.getLogger("googleapiclient.discovery_cache").setLevel(logging.ERROR)
+logging.getLogger("googleapiclient").setLevel(logging.WARNING)
+logging.getLogger("oauth2client").setLevel(logging.WARNING)
+
+# googleapiclient download performs some 3-4 times better with 50 MB chunk size than 5 MB chunk size;
+# but decrypting/decompressing big chunks needs a lot of memory so use smaller chunks on systems with less
+# than 2 GB RAM
+DOWNLOAD_CHUNK_SIZE = 1024 * 1024 * 5 if get_total_memory() < 2048 else 1024 * 1024 * 50
+UPLOAD_CHUNK_SIZE = 1024 * 1024 * 5
+
+
+def get_credentials(credential_file=None, credentials=None):
+    if credential_file:
+        return GoogleCredentials.from_stream(credential_file)
+
+    if credentials and credentials["type"] == "service_account":
+        return ServiceAccountCredentials_from_dict(
+            credentials,
+            scopes=["https://www.googleapis.com/auth/cloud-platform"],
+        )
+
+    if credentials and credentials["type"] == "authorized_user":
+        return GoogleCredentials(
+            access_token=None,
+            client_id=credentials["client_id"],
+            client_secret=credentials["client_secret"],
+            refresh_token=credentials["refresh_token"],
+            token_expiry=None,
+            token_uri=GOOGLE_TOKEN_URI,
+            user_agent="pghoard"
+        )
+
+    return GoogleCredentials.get_application_default()
+
+
+def base64_to_hex(b64val):
+    if isinstance(b64val, str):
+        b64val = b64val.encode("ascii")
+    rawval = codecs.decode(b64val, "base64")
+    hexval = codecs.encode(rawval, "hex")
+    return hexval.decode("ascii")
+
+
+class GoogleTransfer(BaseTransfer):
+    def __init__(self, project_id, bucket_name, credential_file=None, credentials=None, prefix=None, proxy_info=None):
+        super().__init__(prefix=prefix)
+        self.project_id = project_id
+        self.proxy_info = proxy_info
+        self.google_creds = get_credentials(credential_file=credential_file, credentials=credentials)
+        self.gs = self._init_google_client()
+        self.gs_object_client = None
+        self.bucket_name = self.get_or_create_bucket(bucket_name)
+        self.log.debug("GoogleTransfer initialized")
+
+    def _init_google_client(self):
+        start_time = time.monotonic()
+        delay = 2
+        while True:
+            http = build_http()
+            if self.proxy_info:
+                if self.proxy_info.get("type") == "socks5":
+                    proxy_type = httplib2.socks.PROXY_TYPE_SOCKS5
+                else:
+                    proxy_type = httplib2.socks.PROXY_TYPE_HTTP
+
+                http.proxy_info = httplib2.ProxyInfo(
+                    proxy_type,
+                    self.proxy_info["host"],
+                    self.proxy_info["port"],
+                    proxy_user=self.proxy_info.get("user"),
+                    proxy_pass=self.proxy_info.get("pass"),
+                )
+
+            http = self.google_creds.authorize(http)
+
+            try:
+                # sometimes fails: httplib2.ServerNotFoundError: Unable to find the server at www.googleapis.com
+                return build("storage", "v1", http=http)
+            except (httplib2.ServerNotFoundError, socket.timeout):
+                if time.monotonic() - start_time > 600:
+                    raise
+
+            # retry on DNS issues
+            time.sleep(delay)
+            delay = delay * 2
+
+    @contextmanager
+    def _object_client(self, *, not_found=None):
+        """(Re-)initialize object client if required, handle 404 errors gracefully and reset the client on
+        server errors.  Server errors have been shown to be caused by invalid state in the client and do not
+        seem to be resolved without resetting."""
+        if self.gs_object_client is None:
+            if self.gs is None:
+                self.gs = self._init_google_client()
+            self.gs_object_client = self.gs.objects()  # pylint: disable=no-member
+
+        try:
+            yield self.gs_object_client
+        except HttpError as ex:
+            if ex.resp["status"] == "404" and not_found is not None:
+                raise FileNotFoundFromStorageError(not_found)
+            if ex.resp["status"] >= "500" and ex.resp["status"] <= "599":
+                self.log.error("Received server error %r, resetting Google API client", ex.resp["status"])
+                self.gs = None
+                self.gs_object_client = None
+            raise
+
+    def _retry_on_reset(self, request, action):
+        retries = 60
+        retry_wait = 2
+        while True:
+            try:
+                return action()
+            except (IncompleteRead, HttpError, ssl.SSLEOFError, socket.timeout, OSError, socket.gaierror) as ex:
+                # Note that socket.timeout and ssl.SSLEOFError inherit from OSError
+                # and the order of handling the errors here needs to be correct
+                if not retries:
+                    raise
+                elif isinstance(ex, (IncompleteRead, socket.timeout, ssl.SSLEOFError, BrokenPipeError)):
+                    pass  # just retry with the same sleep amount
+                elif isinstance(ex, HttpError):
+                    # https://cloud.google.com/storage/docs/json_api/v1/status-codes
+                    # https://cloud.google.com/storage/docs/exponential-backoff
+                    if ex.resp["status"] not in ("429", "500", "502", "503", "504"):  # pylint: disable=no-member
+                        raise
+                    retry_wait = min(10.0, max(1.0, retry_wait * 2) + random.random())
+                # httplib2 commonly fails with Bad File Descriptor and Connection Reset
+                elif isinstance(ex, OSError) and ex.errno not in [errno.EAGAIN, errno.EBADF, errno.ECONNRESET]:
+                    raise
+                # getaddrinfo sometimes fails with "Name or service not known"
+                elif isinstance(ex, socket.gaierror) and ex.errno != socket.EAI_NONAME:
+                    raise
+
+                self.log.warning("%s failed: %s (%s), retrying in %.2fs", action, ex.__class__.__name__, ex, retry_wait)
+
+            # we want to reset the http connection state in case of error
+            if request and hasattr(request, "http"):
+                request.http.connections.clear()  # reset connection cache
+
+            retries -= 1
+            time.sleep(retry_wait)
+
+    def copy_file(self, *, source_key, destination_key, metadata=None, **_kwargs):
+        source_object = self.format_key_for_backend(source_key)
+        destination_object = self.format_key_for_backend(destination_key)
+        body = {}
+        if metadata is not None:
+            body["metadata"] = metadata
+
+        with self._object_client(not_found=source_key) as clob:
+            request = clob.copy(
+                body=body,
+                destinationBucket=self.bucket_name,
+                destinationObject=destination_object,
+                sourceBucket=self.bucket_name,
+                sourceObject=source_object,
+            )
+            self._retry_on_reset(request, request.execute)
+
+    def get_metadata_for_key(self, key):
+        key = self.format_key_for_backend(key)
+        with self._object_client(not_found=key) as clob:
+            return self._metadata_for_key(clob, key)
+
+    def _metadata_for_key(self, clob, key):
+        req = clob.get(bucket=self.bucket_name, object=key)
+        obj = self._retry_on_reset(req, req.execute)
+        return obj.get("metadata", {})
+
+    def _unpaginate(self, domain, initial_op, *, on_properties):
+        """Iterate thru the request pages until all items have been processed"""
+        request = initial_op(domain)
+        while request is not None:
+            result = self._retry_on_reset(request, request.execute)
+            for on_property in on_properties:
+                items = result.get(on_property)
+                if items is not None:
+                    yield on_property, items
+            request = domain.list_next(request, result)
+
+    def iter_key(
+        self,
+        key,
+        *,
+        with_metadata=True,  # pylint: disable=unused-argument, unused-variable
+        deep=False,
+        include_key=False
+    ):
+        path = self.format_key_for_backend(key, trailing_slash=not include_key)
+        self.log.debug("Listing path %r", path)
+        with self._object_client() as clob:
+
+            def initial_op(domain):
+                if deep:
+                    kwargs = {}
+                else:
+                    kwargs = {"delimiter": "/"}
+                return domain.list(bucket=self.bucket_name, prefix=path, **kwargs)
+
+            for property_name, items in self._unpaginate(clob, initial_op, on_properties=["items", "prefixes"]):
+                if property_name == "items":
+                    for item in items:
+                        if item["name"].endswith("/"):
+                            self.log.warning("list_iter: directory entry %r", item)
+                            continue  # skip directory level objects
+
+                        yield IterKeyItem(
+                            type=KEY_TYPE_OBJECT,
+                            value={
+                                "name": self.format_key_from_backend(item["name"]),
+                                "size": int(item["size"]),
+                                "last_modified": parse_timestamp(item["updated"]),
+                                "metadata": item.get("metadata", {}),
+                                "md5": base64_to_hex(item["md5Hash"]),
+                            },
+                        )
+                elif property_name == "prefixes":
+                    for prefix in items:
+                        yield IterKeyItem(type=KEY_TYPE_PREFIX, value=self.format_key_from_backend(prefix).rstrip("/"))
+                else:
+                    raise NotImplementedError(property_name)
+
+    def delete_key(self, key):
+        key = self.format_key_for_backend(key)
+        self.log.debug("Deleting key: %r", key)
+        with self._object_client(not_found=key) as clob:
+            req = clob.delete(bucket=self.bucket_name, object=key)
+            self._retry_on_reset(req, req.execute)
+
+    def get_contents_to_file(self, key, filepath_to_store_to, *, progress_callback=None):
+        fileobj = FileIO(filepath_to_store_to, mode="wb")
+        done = False
+        metadata = {}
+        try:
+            metadata = self.get_contents_to_fileobj(key, fileobj, progress_callback=progress_callback)
+            done = True
+        finally:
+            fileobj.close()
+            if not done:
+                os.unlink(filepath_to_store_to)
+        return metadata
+
+    def get_contents_to_fileobj(self, key, fileobj_to_store_to, *, progress_callback=None):
+        key = self.format_key_for_backend(key)
+        self.log.debug("Starting to fetch the contents of: %r to %r", key, fileobj_to_store_to)
+        next_prog_report = 0.0
+        last_log_output = 0.0
+        with self._object_client(not_found=key) as clob:
+            req = clob.get_media(bucket=self.bucket_name, object=key)
+            download = MediaIoBaseDownload(fileobj_to_store_to, req, chunksize=DOWNLOAD_CHUNK_SIZE)
+            done = False
+            while not done:
+                status, done = self._retry_on_reset(getattr(download, "_request", None), download.next_chunk)
+                if status:
+                    progress_pct = status.progress() * 100
+                    now = time.monotonic()
+                    if (now - last_log_output) >= 5.0:
+                        self.log.debug("Download of %r: %d%%", key, progress_pct)
+                        last_log_output = now
+
+                    if progress_callback and progress_pct > next_prog_report:
+                        progress_callback(progress_pct, 100)
+                        next_prog_report = progress_pct + 0.1
+            return self._metadata_for_key(clob, key)
+
+    def get_contents_to_string(self, key):
+        key = self.format_key_for_backend(key)
+        self.log.debug("Starting to fetch the contents of: %r", key)
+        with self._object_client(not_found=key) as clob:
+            req = clob.get_media(bucket=self.bucket_name, object=key)
+            data = self._retry_on_reset(req, req.execute)
+            return data, self._metadata_for_key(clob, key)
+
+    def get_file_size(self, key):
+        key = self.format_key_for_backend(key)
+        with self._object_client(not_found=key) as clob:
+            req = clob.get(bucket=self.bucket_name, object=key)
+            obj = self._retry_on_reset(req, req.execute)
+            return int(obj["size"])
+
+    def _upload(self, upload, key, metadata, extra_props, cache_control, upload_progress_fn=None):
+        key = self.format_key_for_backend(key)
+        self.log.debug("Starting to upload %r", key)
+        body = {"metadata": metadata}
+        if extra_props:
+            body.update(extra_props)
+        if cache_control is not None:
+            body["cacheControl"] = cache_control
+
+        last_log_output = 0.0
+        with self._object_client() as clob:
+            req = clob.insert(bucket=self.bucket_name, name=key, media_body=upload, body=body)
+            response = None
+            while response is None:
+                status, response = self._retry_on_reset(req, req.next_chunk)
+                if status:
+                    now = time.monotonic()
+                    if (now - last_log_output) >= 5.0:
+                        self.log.debug(
+                            "Upload of %r to %r: %d%%, %s bytes", upload, key,
+                            status.progress() * 100, status.resumable_progress
+                        )
+                        last_log_output = now
+
+                    if upload_progress_fn:
+                        upload_progress_fn(status.resumable_progress)
+
+    # pylint: disable=arguments-differ
+    def store_file_from_memory(self, key, memstring, metadata=None, extra_props=None, cache_control=None, mimetype=None):
+        upload = MediaIoBaseUpload(
+            BytesIO(memstring), mimetype or "application/octet-stream", chunksize=UPLOAD_CHUNK_SIZE, resumable=True
+        )
+        return self._upload(upload, key, self.sanitize_metadata(metadata), extra_props, cache_control=cache_control)
+
+    # pylint: disable=arguments-differ
+    def store_file_from_disk(
+        self,
+        key,
+        filepath,
+        metadata=None,  # pylint: disable=arguments-differ, unused-variable
+        *,
+        multipart=None,
+        extra_props=None,  # pylint: disable=arguments-differ, unused-variable
+        cache_control=None,
+        mimetype=None
+    ):
+        mimetype = mimetype or "application/octet-stream"
+        upload = MediaFileUpload(filepath, mimetype, chunksize=UPLOAD_CHUNK_SIZE, resumable=True)
+        return self._upload(upload, key, self.sanitize_metadata(metadata), extra_props, cache_control=cache_control)
+
+    def store_file_object(self, key, fd, *, cache_control=None, metadata=None, mimetype=None, upload_progress_fn=None):
+        mimetype = mimetype or "application/octet-stream"
+        return self._upload(
+            MediaStreamUpload(fd, chunk_size=UPLOAD_CHUNK_SIZE, mime_type=mimetype, name=key),
+            key,
+            self.sanitize_metadata(metadata),
+            None,
+            cache_control=cache_control,
+            upload_progress_fn=upload_progress_fn,
+        )
+
+    def get_or_create_bucket(self, bucket_name):
+        """Look up the bucket if it already exists and try to create the
+        bucket in case it doesn't.  Note that we can't just always try to
+        unconditionally create the bucket as Google imposes a strict rate
+        limit on bucket creation operations, even if it doesn't result in a
+        new bucket.
+
+        Quietly handle the case where the bucket already exists to avoid
+        race conditions.  Note that we'll get a 400 Bad Request response for
+        invalid bucket names ("Invalid bucket name") as well as for invalid
+        project ("Invalid argument"), try to handle both gracefully."""
+        start_time = time.time()
+        gs_buckets = self.gs.buckets()  # pylint: disable=no-member
+        try:
+            request = gs_buckets.get(bucket=bucket_name)
+            self._retry_on_reset(request, request.execute)
+            self.log.debug("Bucket: %r already exists, took: %.3fs", bucket_name, time.time() - start_time)
+        except HttpError as ex:
+            if ex.resp["status"] == "404":
+                pass  # we need to create it
+            elif ex.resp["status"] == "403":
+                raise InvalidConfigurationError("Bucket {0!r} exists but isn't accessible".format(bucket_name))
+            else:
+                raise
+        else:
+            return bucket_name
+
+        try:
+            req = gs_buckets.insert(project=self.project_id, body={"name": bucket_name})
+            self._retry_on_reset(req, req.execute)
+            self.log.debug("Created bucket: %r successfully, took: %.3fs", bucket_name, time.time() - start_time)
+        except HttpError as ex:
+            error = json.loads(ex.content.decode("utf-8"))["error"]
+            if error["message"].startswith("You already own this bucket"):
+                self.log.debug("Bucket: %r already exists, took: %.3fs", bucket_name, time.time() - start_time)
+            elif error["message"] == "Invalid argument.":
+                raise InvalidConfigurationError("Invalid project id {0!r}".format(self.project_id))
+            elif error["message"].startswith("Invalid bucket name"):
+                raise InvalidConfigurationError("Invalid bucket name {0!r}".format(bucket_name))
+            else:
+                raise
+
+        return bucket_name
+
+
+class MediaStreamUpload(MediaUpload):
+    """Support streaming arbitrary amount of data from non-seekable object supporting read method."""
+    def __init__(self, fd, *, chunk_size, mime_type, name):
+        self._data = b""
+        self._next_chunk = b""
+        self._chunk_size = chunk_size
+        self._fd = fd
+        self._mime_type = mime_type
+        self._name = name
+        self._position = None
+
+    def chunksize(self):
+        return self._chunk_size
+
+    def mimetype(self):
+        return self._mime_type
+
+    def size(self):
+        self.peek()
+        if len(self._next_chunk) < self.peeksize:
+            # The total file size should be returned if we have hit the final chunk.
+            return (self._position or 0) + len(self._data) + len(self._next_chunk)
+        return None
+
+    def resumable(self):
+        return True
+
+    @property
+    def peeksize(self):
+        # Using 1 extra byte to avoid perfectly aligned file
+        return self._chunk_size + 1
+
+    def peek(self):
+        """try to top up some data into _next_chunk"""
+        if len(self._next_chunk) < self.peeksize:
+            # top-up the _next_chunk
+            self._next_chunk = self._read_bytes(self.peeksize - len(self._next_chunk), initial_data=self._next_chunk)
+
+    # second parameter is length but baseclass incorrectly names it end
+    def getbytes(self, begin, length):  # pylint: disable=arguments-differ
+        if begin < (self._position or 0):
+            msg = "Requested position {} for {!r} preceeds already fulfilled position {}".format(
+                begin, self._name, self._position
+            )
+            raise IndexError(msg)
+        elif begin > (self._position or 0) + len(self._data):
+            msg = "Requested position {} for {!r} has gap from previous position {} and {} byte chunk".format(
+                begin, self._name, self._position, len(self._data)
+            )
+            raise IndexError(msg)
+
+        if self._position is None or begin == self._position + len(self._data):
+            if length <= len(self._next_chunk):
+                self._data = self._next_chunk[:length]
+                self._next_chunk = self._next_chunk[length:]
+            else:
+                self._data = self._read_bytes(length - len(self._next_chunk), initial_data=self._next_chunk)
+                self._next_chunk = b""
+        elif begin != self._position or length > len(self._data):
+            retain_chunk = self._data[begin - self._position:]
+            bytes_remaining = length - len(retain_chunk)
+            if 0 < bytes_remaining <= len(self._next_chunk):
+                self._data = retain_chunk + self._next_chunk[:bytes_remaining]
+                self._next_chunk = self._next_chunk[bytes_remaining:]
+            elif bytes_remaining > len(self._next_chunk):
+                retain_chunk += self._next_chunk
+                self._next_chunk = b""
+                self._data = self._read_bytes(length - len(retain_chunk), initial_data=retain_chunk)
+
+        self.peek()
+        self._position = begin
+        return self._data
+
+    def has_stream(self):
+        return False
+
+    def stream(self):
+        raise NotImplementedError
+
+    def _read_bytes(self, length, *, initial_data=None):
+        bytes_remaining = length
+        read_results = []
+        if initial_data:
+            read_results.append(initial_data)
+        while bytes_remaining > 0:
+            data = self._fd.read(bytes_remaining)
+            if data:
+                read_results.append(data)
+                bytes_remaining -= len(data)
+            else:
+                break
+
+        if not read_results:
+            return b""
+        elif len(read_results) == 1:
+            return read_results[0]
+        else:
+            return b"".join(read_results)
diff --git a/rohmu/object_storage/local.py b/rohmu/object_storage/local.py
new file mode 100644
index 00000000..b5f6b047
--- /dev/null
+++ b/rohmu/object_storage/local.py
@@ -0,0 +1,238 @@
+"""
+rohmu - local filesystem interface
+
+Copyright (c) 2016 Ohmu Ltd
+See LICENSE for details
+"""
+import contextlib
+import datetime
+import json
+import os
+import shutil
+import tempfile
+from io import BytesIO
+
+from ..errors import FileNotFoundFromStorageError, LocalFileIsRemoteFileError
+from .base import KEY_TYPE_OBJECT, KEY_TYPE_PREFIX, BaseTransfer, IterKeyItem
+
+CHUNK_SIZE = 1024 * 1024
+
+
+class LocalTransfer(BaseTransfer):
+    def __init__(self, directory, prefix=None):
+        prefix = os.path.join(directory, (prefix or "").strip("/"))
+        super().__init__(prefix=prefix)
+        self.log.debug("LocalTransfer initialized")
+
+    def copy_file(self, *, source_key, destination_key, metadata=None, **_kwargs):
+        source_path = self.format_key_for_backend(source_key.strip("/"))
+        destination_path = self.format_key_for_backend(destination_key.strip("/"))
+        if not os.path.isfile(source_path):
+            raise FileNotFoundFromStorageError(source_key)
+        os.makedirs(os.path.dirname(destination_path), exist_ok=True)
+        shutil.copy(source_path, destination_path)
+        if metadata is None:
+            shutil.copy(source_path + ".metadata", destination_path + ".metadata")
+        else:
+            self._save_metadata(destination_path, metadata)
+
+    def get_metadata_for_key(self, key):
+        source_path = self.format_key_for_backend(key.strip("/"))
+        if not os.path.exists(source_path):
+            raise FileNotFoundFromStorageError(key)
+        metadata_path = source_path + ".metadata"
+        try:
+            with open(metadata_path, "r") as fp:
+                return json.load(fp)
+        except FileNotFoundError:
+            raise FileNotFoundFromStorageError(key)
+
+    def delete_key(self, key):
+        self.log.debug("Deleting key: %r", key)
+        target_path = self.format_key_for_backend(key.strip("/"))
+        if not os.path.exists(target_path):
+            raise FileNotFoundFromStorageError(key)
+        os.unlink(target_path)
+        metadata_tmp_path = target_path + ".metadata_tmp"
+        with contextlib.suppress(FileNotFoundError):
+            os.unlink(metadata_tmp_path)
+        metadata_path = target_path + ".metadata"
+        with contextlib.suppress(FileNotFoundError):
+            os.unlink(metadata_path)
+
+    def delete_tree(self, key):
+        self.log.debug("Deleting tree: %r", key)
+        target_path = self.format_key_for_backend(key.strip("/"))
+        if not os.path.isdir(target_path):
+            raise FileNotFoundFromStorageError(key)
+        shutil.rmtree(target_path)
+
+    @staticmethod
+    def _skip_file_name(file_name):
+        return file_name.startswith(".") or file_name.endswith(".metadata") or ".metadata_tmp" in file_name
+
+    @staticmethod
+    def _yield_object(key, full_path, with_metadata):
+        metadata_file = full_path + ".metadata"
+        if not os.path.exists(metadata_file):
+            return
+        if with_metadata:
+            with open(metadata_file, "r") as fp:
+                metadata = json.load(fp)
+        else:
+            metadata = None
+        st = os.stat(full_path)
+        last_modified = datetime.datetime.fromtimestamp(st.st_mtime, tz=datetime.timezone.utc)
+        yield IterKeyItem(
+            type=KEY_TYPE_OBJECT,
+            value={
+                "name": key,
+                "size": st.st_size,
+                "last_modified": last_modified,
+                "metadata": metadata,
+            },
+        )
+
+    def iter_key(self, key, *, with_metadata=True, deep=False, include_key=False):
+        target_path = self.format_key_for_backend(key.strip("/"))
+        try:
+            input_files = os.listdir(target_path)
+        except FileNotFoundError:
+            return
+        except NotADirectoryError:
+            if include_key:
+                file_name = os.path.basename(target_path)
+                if self._skip_file_name(file_name):
+                    return
+                yield from self._yield_object(key.strip("/"), target_path, with_metadata=with_metadata)
+            return
+
+        for file_name in input_files:
+            if self._skip_file_name(file_name):
+                continue
+            full_path = os.path.join(target_path, file_name)
+            if os.path.isdir(full_path):
+                file_key = os.path.join(key.strip("/"), file_name)
+                if deep:
+                    yield from self.iter_key(file_key, with_metadata=with_metadata, deep=True)
+                else:
+                    yield IterKeyItem(type=KEY_TYPE_PREFIX, value=file_key)
+            else:
+                # Don't return files if metadata file is not present; files are written in two phases and
+                # should be considered available only after also metadata has been written
+                if not os.path.exists(full_path + ".metadata"):
+                    continue
+                yield from self._yield_object(
+                    key=os.path.join(key.strip("/"), file_name),
+                    full_path=full_path,
+                    with_metadata=with_metadata,
+                )
+
+    def get_contents_to_file(self, key, filepath_to_store_to, *, progress_callback=None):
+        source_path = self.format_key_for_backend(key.strip("/"))
+        try:
+            src_stat = os.stat(source_path)
+        except FileNotFoundError:
+            raise FileNotFoundFromStorageError(key)
+        with contextlib.suppress(FileNotFoundError):
+            dst_stat = os.stat(filepath_to_store_to)
+            if dst_stat.st_dev == src_stat.st_dev and dst_stat.st_ino == src_stat.st_ino:
+                raise LocalFileIsRemoteFileError(source_path)
+        with open(filepath_to_store_to, "wb") as fileobj_to_store_to:
+            return self.get_contents_to_fileobj(key, fileobj_to_store_to, progress_callback=progress_callback)
+
+    def get_contents_to_fileobj(self, key, fileobj_to_store_to, *, progress_callback=None):
+        source_path = self.format_key_for_backend(key.strip("/"))
+        if not os.path.exists(source_path):
+            raise FileNotFoundFromStorageError(key)
+
+        input_size = os.stat(source_path).st_size
+        bytes_written = 0
+        with open(source_path, "rb") as fp:
+            while True:
+                buf = fp.read(CHUNK_SIZE)
+                if not buf:
+                    break
+                fileobj_to_store_to.write(buf)
+                bytes_written += len(buf)
+                if progress_callback:
+                    progress_callback(bytes_written, input_size)
+
+        return self.get_metadata_for_key(key)
+
+    def get_contents_to_string(self, key):
+        bio = BytesIO()
+        metadata = self.get_contents_to_fileobj(key, bio)
+        return bio.getvalue(), metadata
+
+    def get_file_size(self, key):
+        source_path = self.format_key_for_backend(key.strip("/"))
+        if not os.path.exists(source_path):
+            raise FileNotFoundFromStorageError(key)
+        return os.stat(source_path).st_size
+
+    def _save_metadata(self, target_path, metadata):
+        metadata_path = target_path + ".metadata"
+        with atomic_create_file(metadata_path) as fp:
+            json.dump(self.sanitize_metadata(metadata), fp)
+
+    def store_file_from_memory(self, key, memstring, metadata=None, cache_control=None, mimetype=None):
+        target_path = self.format_key_for_backend(key.strip("/"))
+        os.makedirs(os.path.dirname(target_path), exist_ok=True)
+        with open(target_path, "wb") as fp:
+            fp.write(memstring)
+        self._save_metadata(target_path, metadata)
+
+    def store_file_from_disk(self, key, filepath, metadata=None, multipart=None, cache_control=None, mimetype=None):
+        target_path = self.format_key_for_backend(key.strip("/"))
+        src_stat = os.stat(filepath)
+        with contextlib.suppress(FileNotFoundError):
+            dst_stat = os.stat(target_path)
+            if dst_stat.st_dev == src_stat.st_dev and dst_stat.st_ino == src_stat.st_ino:
+                self._save_metadata(target_path, metadata)
+                raise LocalFileIsRemoteFileError(target_path)
+        os.makedirs(os.path.dirname(target_path), exist_ok=True)
+        shutil.copyfile(filepath, target_path)
+        self._save_metadata(target_path, metadata)
+
+    def store_file_object(
+        self,
+        key,
+        fd,
+        *,
+        cache_control=None,  # pylint: disable=unused-argument
+        metadata=None,
+        mimetype=None,
+        upload_progress_fn=None
+    ):  # pylint: disable=unused-argument
+        target_path = self.format_key_for_backend(key.strip("/"))
+        os.makedirs(os.path.dirname(target_path), exist_ok=True)
+        bytes_written = 0
+        with open(target_path, "wb") as output_fp:
+            while True:
+                data = fd.read(1024 * 1024)
+                if not data:
+                    break
+                output_fp.write(data)
+                bytes_written += len(data)
+                if upload_progress_fn:
+                    upload_progress_fn(bytes_written)
+
+        self._save_metadata(target_path, metadata)
+
+
+@contextlib.contextmanager
+def atomic_create_file(file_path):
+    """Open a temporary file for writing, rename to final name when done"""
+    fd, tmp_file_path = tempfile.mkstemp(
+        prefix=os.path.basename(file_path), dir=os.path.dirname(file_path), suffix=".metadata_tmp"
+    )
+    try:
+        with os.fdopen(fd, "w") as out_file:
+            yield out_file
+
+        os.rename(tmp_file_path, file_path)
+    except Exception:  # pytest: disable=broad-except
+        with contextlib.suppress(Exception):
+            os.unlink(tmp_file_path)
+        raise
diff --git a/rohmu/object_storage/s3.py b/rohmu/object_storage/s3.py
new file mode 100644
index 00000000..d1ad5ed0
--- /dev/null
+++ b/rohmu/object_storage/s3.py
@@ -0,0 +1,453 @@
+"""
+rohmu
+
+Copyright (c) 2016 Ohmu Ltd
+See LICENSE for details
+"""
+import math
+import os
+import time
+
+import botocore.client
+import botocore.config
+import botocore.exceptions
+import botocore.session
+
+from ..errors import (FileNotFoundFromStorageError, InvalidConfigurationError, StorageError)
+from .base import (KEY_TYPE_OBJECT, KEY_TYPE_PREFIX, BaseTransfer, IterKeyItem, get_total_memory)
+
+
+def calculate_chunk_size():
+    total_mem_mib = get_total_memory() or 0
+    # At least 5 MiB, at most 524 MiB. Max block size used for hosts with ~210+ GB of memory
+    return max(min(int(total_mem_mib / 400), 524), 5) * 1024 * 1024
+
+
+def get_proxy_url(proxy_info):
+    username = proxy_info.get("user")
+    password = proxy_info.get("pass")
+    if username and password:
+        auth = f"{username}:{password}@"
+    else:
+        auth = ""
+    host = proxy_info["host"]
+    port = proxy_info["port"]
+    if proxy_info.get("type") in {"socks5", "socks5h"}:
+        schema = proxy_info.get("type")
+    else:
+        schema = "http"
+    proxy_url = f"{schema}://{auth}{host}:{port}"
+    return proxy_url
+
+
+# Set chunk size based on host memory. S3 supports up to 10k chunks and up to 5 TiB individual
+# files. Minimum chunk size is 5 MiB, which means max ~50 GB files can be uploaded. In order to get
+# to that 5 TiB increase the block size based on host memory; we don't want to use the max for all
+# hosts to avoid allocating too large portion of all available memory.
+MULTIPART_CHUNK_SIZE = calculate_chunk_size()
+READ_BLOCK_SIZE = 1024 * 1024 * 1
+
+
+class S3Transfer(BaseTransfer):
+    def __init__(
+        self,
+        region,
+        bucket_name,
+        aws_access_key_id=None,
+        aws_secret_access_key=None,
+        prefix=None,
+        host=None,
+        port=None,
+        is_secure=False,
+        is_verify_tls=False,
+        segment_size=MULTIPART_CHUNK_SIZE,
+        encrypted=False,
+        proxy_info=None
+    ):
+        super().__init__(prefix=prefix)
+        botocore_session = botocore.session.get_session()
+        self.bucket_name = bucket_name
+        self.location = ""
+        self.region = region
+        if not host or not port:
+            custom_config = {}
+            if proxy_info:
+                proxy_url = get_proxy_url(proxy_info)
+                custom_config["proxies"] = {"https": proxy_url}
+            self.s3_client = botocore_session.create_client(
+                "s3",
+                config=botocore.config.Config(**custom_config),
+                aws_access_key_id=aws_access_key_id,
+                aws_secret_access_key=aws_secret_access_key,
+                region_name=region,
+            )
+            if self.region and self.region != "us-east-1":
+                self.location = self.region
+        else:
+            scheme = "https" if is_secure else "http"
+            custom_url = "{scheme}://{host}:{port}".format(scheme=scheme, host=host, port=port)
+            if self.region:
+                signature_version = "s3v4"
+                self.location = self.region
+            else:
+                signature_version = "s3"
+            proxies = {}
+            if proxy_info:
+                proxy_url = get_proxy_url(proxy_info)
+                proxies["proxies"] = {"https": proxy_url}
+            custom_config = botocore.client.Config(
+                s3={"addressing_style": "path"},
+                signature_version=signature_version,
+                **proxies,
+            )
+            self.s3_client = botocore_session.create_client(
+                "s3",
+                aws_access_key_id=aws_access_key_id,
+                aws_secret_access_key=aws_secret_access_key,
+                config=custom_config,
+                endpoint_url=custom_url,
+                region_name=region,
+                verify=is_verify_tls,
+            )
+
+        self.check_or_create_bucket()
+
+        self.multipart_chunk_size = segment_size
+        self.encrypted = encrypted
+        self.log.debug("S3Transfer initialized")
+
+    def copy_file(self, *, source_key, destination_key, metadata=None, **_kwargs):
+        source_path = self.bucket_name + "/" + self.format_key_for_backend(source_key, remove_slash_prefix=True)
+        destination_path = self.format_key_for_backend(destination_key, remove_slash_prefix=True)
+        try:
+            self.s3_client.copy_object(
+                Bucket=self.bucket_name,
+                CopySource=source_path,
+                Key=destination_path,
+                Metadata=metadata or {},
+                MetadataDirective="COPY" if metadata is None else "REPLACE",
+            )
+        except botocore.exceptions.ClientError as ex:
+            status_code = ex.response.get("ResponseMetadata", {}).get("HTTPStatusCode")
+            if status_code == 404:
+                raise FileNotFoundFromStorageError(source_key)
+            else:
+                raise StorageError("Copying {!r} to {!r} failed: {!r}".format(source_key, destination_key, ex)) from ex
+
+    def get_metadata_for_key(self, key):
+        key = self.format_key_for_backend(key, remove_slash_prefix=True)
+        return self._metadata_for_key(key)
+
+    def _metadata_for_key(self, key):
+        try:
+            response = self.s3_client.head_object(Bucket=self.bucket_name, Key=key)
+        except botocore.exceptions.ClientError as ex:
+            status_code = ex.response.get("ResponseMetadata", {}).get("HTTPStatusCode")
+            if status_code == 404:
+                raise FileNotFoundFromStorageError(key)
+            else:
+                raise StorageError("Metadata lookup failed for {}".format(key)) from ex
+
+        return response["Metadata"]
+
+    def delete_key(self, key):
+        key = self.format_key_for_backend(key, remove_slash_prefix=True)
+        self.log.debug("Deleting key: %r", key)
+        self._metadata_for_key(key)  # check that key exists
+        self.s3_client.delete_object(Bucket=self.bucket_name, Key=key)
+
+    def delete_tree(self, key):
+        key = self.format_key_for_backend(key, remove_slash_prefix=True, trailing_slash=True)
+        self.log.debug("Deleting tree: %r", key)
+        objects_to_delete = self.s3_client.list_objects(Bucket=self.bucket_name, Prefix=key)
+        delete_keys = [{"Key": key} for key in [obj["Key"] for obj in objects_to_delete.get("Contents", [])]]
+        if delete_keys:
+            self.s3_client.delete_objects(Bucket=self.bucket_name, Delete={"Objects": delete_keys})
+
+    def iter_key(self, key, *, with_metadata=True, deep=False, include_key=False):
+        path = self.format_key_for_backend(key, remove_slash_prefix=True, trailing_slash=not include_key)
+        self.log.debug("Listing path %r", path)
+        continuation_token = None
+        while True:
+            args = {
+                "Bucket": self.bucket_name,
+                "Prefix": path,
+            }
+            if not deep:
+                args["Delimiter"] = "/"
+            if continuation_token:
+                args["ContinuationToken"] = continuation_token
+            response = self.s3_client.list_objects_v2(**args)
+
+            for item in response.get("Contents", []):
+                if with_metadata:
+                    metadata = {k.lower(): v for k, v in self._metadata_for_key(item["Key"]).items()}
+                else:
+                    metadata = None
+                name = self.format_key_from_backend(item["Key"])
+                yield IterKeyItem(
+                    type=KEY_TYPE_OBJECT,
+                    value={
+                        "last_modified": item["LastModified"],
+                        "md5": item["ETag"].strip('"'),
+                        "metadata": metadata,
+                        "name": name,
+                        "size": item["Size"],
+                    },
+                )
+
+            for common_prefix in response.get("CommonPrefixes", []):
+                yield IterKeyItem(
+                    type=KEY_TYPE_PREFIX,
+                    value=self.format_key_from_backend(common_prefix["Prefix"]).rstrip("/"),
+                )
+
+            if "NextContinuationToken" in response:
+                continuation_token = response["NextContinuationToken"]
+            else:
+                break
+
+    def _get_object_stream(self, key):
+        key = self.format_key_for_backend(key, remove_slash_prefix=True)
+        try:
+            response = self.s3_client.get_object(
+                Bucket=self.bucket_name,
+                Key=key,
+            )
+        except botocore.exceptions.ClientError as ex:
+            status_code = ex.response.get("ResponseMetadata", {}).get("HTTPStatusCode")
+            if status_code == 404:
+                raise FileNotFoundFromStorageError(key)
+            else:
+                raise StorageError("Fetching the remote object {} failed".format(key)) from ex
+        return response["Body"], response["ContentLength"], response["Metadata"]
+
+    def _read_object_to_fileobj(self, fileobj, streaming_body, body_length, cb=None):
+        data_read = 0
+        while data_read < body_length:
+            read_amount = body_length - data_read
+            if read_amount > READ_BLOCK_SIZE:
+                read_amount = READ_BLOCK_SIZE
+            data = streaming_body.read(amt=read_amount)
+            fileobj.write(data)
+            data_read += len(data)
+            if cb:
+                cb(data_read, body_length)
+        if cb:
+            cb(data_read, body_length)
+
+    def get_contents_to_file(self, key, filepath_to_store_to, *, progress_callback=None):
+        with open(filepath_to_store_to, "wb") as fh:
+            stream, length, metadata = self._get_object_stream(key)
+            self._read_object_to_fileobj(fh, stream, length, cb=progress_callback)
+        return metadata
+
+    def get_contents_to_fileobj(self, key, fileobj_to_store_to, *, progress_callback=None):
+        stream, length, metadata = self._get_object_stream(key)
+        self._read_object_to_fileobj(fileobj_to_store_to, stream, length, cb=progress_callback)
+        return metadata
+
+    def get_contents_to_string(self, key):
+        stream, _, metadata = self._get_object_stream(key)
+        data = stream.read()
+        return data, metadata
+
+    def get_file_size(self, key):
+        key = self.format_key_for_backend(key, remove_slash_prefix=True)
+        try:
+            response = self.s3_client.head_object(Bucket=self.bucket_name, Key=key)
+            return int(response["ContentLength"])
+        except botocore.exceptions.ClientError as ex:
+            if ex.response.get("ResponseMetadata", {}).get("HTTPStatusCode") == 404:
+                raise FileNotFoundFromStorageError(key)
+            else:
+                raise StorageError("File size lookup failed for {}".format(key)) from ex
+
+    def store_file_from_memory(self, key, memstring, metadata=None, cache_control=None, mimetype=None):
+        key = self.format_key_for_backend(key, remove_slash_prefix=True)
+        args = {
+            "Bucket": self.bucket_name,
+            "Body": bytes(memstring),  # make sure Body is of type bytes as memoryview's not allowed, only bytes/bytearrays
+            "Key": key,
+        }
+        if metadata:
+            args["Metadata"] = self.sanitize_metadata(metadata)
+        if self.encrypted:
+            args["ServerSideEncryption"] = "AES256"
+        if cache_control is not None:
+            args["CacheControl"] = cache_control
+        if mimetype is not None:
+            args["ContentType"] = mimetype
+        self.s3_client.put_object(**args)
+
+    def store_file_from_disk(self, key, filepath, metadata=None, multipart=None, cache_control=None, mimetype=None):
+        size = os.path.getsize(filepath)
+        if not multipart or size <= self.multipart_chunk_size:
+            with open(filepath, "rb") as fh:
+                data = fh.read()
+                self.store_file_from_memory(key, data, metadata, cache_control=cache_control)
+            return
+
+        with open(filepath, "rb") as fp:
+            self.multipart_upload_file_object(
+                cache_control=cache_control, fp=fp, key=key, metadata=metadata, mimetype=mimetype, size=size
+            )
+
+    def multipart_upload_file_object(self, *, cache_control, fp, key, metadata, mimetype, progress_fn=None, size=None):
+        key = self.format_key_for_backend(key, remove_slash_prefix=True)
+        start_of_multipart_upload = time.monotonic()
+        bytes_sent = 0
+
+        chunks = "Unknown"
+        if size is not None:
+            chunks = math.ceil(size / self.multipart_chunk_size)
+        self.log.debug("Starting to upload multipart file: %r, size: %s, chunks: %s", key, size, chunks)
+
+        parts = []
+        part_number = 1
+
+        args = {
+            "Bucket": self.bucket_name,
+            "Key": key,
+        }
+        if metadata:
+            args["Metadata"] = self.sanitize_metadata(metadata)
+        if self.encrypted:
+            args["ServerSideEncryption"] = "AES256"
+        if cache_control is not None:
+            args["CacheControl"] = cache_control
+        if mimetype is not None:
+            args["ContentType"] = mimetype
+        try:
+            response = self.s3_client.create_multipart_upload(**args)
+        except botocore.exceptions.ClientError as ex:
+            raise StorageError("Failed to initiate multipart upload for {}".format(key)) from ex
+
+        mp_id = response["UploadId"]
+
+        while True:
+            data = self._read_bytes(fp, self.multipart_chunk_size)
+            if not data:
+                break
+
+            attempts = 10
+            start_of_part_upload = time.monotonic()
+            while True:
+                attempts -= 1
+                try:
+                    response = self.s3_client.upload_part(
+                        Body=data,
+                        Bucket=self.bucket_name,
+                        Key=key,
+                        PartNumber=part_number,
+                        UploadId=mp_id,
+                    )
+                except botocore.exceptions.ClientError as ex:
+                    self.log.exception("Uploading part %d for %s failed, attempts left: %d", part_number, key, attempts)
+                    if attempts <= 0:
+                        try:
+                            self.s3_client.abort_multipart_upload(
+                                Bucket=self.bucket_name,
+                                Key=key,
+                                UploadId=mp_id,
+                            )
+                        finally:
+                            err = "Multipart upload of {0} failed: {1.__class__.__name__}: {1}".format(key, ex)
+                            raise StorageError(err) from ex
+                    else:
+                        time.sleep(1.0)
+                else:
+                    self.log.info(
+                        "Uploaded part %s of %s, size %s in %.2fs",
+                        part_number,
+                        chunks,
+                        len(data),
+                        time.monotonic() - start_of_part_upload,
+                    )
+                    parts.append({
+                        "ETag": response["ETag"],
+                        "PartNumber": part_number,
+                    })
+                    part_number += 1
+                    bytes_sent += len(data)
+                    if progress_fn:
+                        progress_fn(bytes_sent)
+                    break
+
+        try:
+            self.s3_client.complete_multipart_upload(
+                Bucket=self.bucket_name,
+                Key=key,
+                MultipartUpload={"Parts": parts},
+                UploadId=mp_id,
+            )
+        except botocore.exceptions.ClientError as ex:
+            try:
+                self.s3_client.abort_multipart_upload(
+                    Bucket=self.bucket_name,
+                    Key=key,
+                    UploadId=mp_id,
+                )
+            finally:
+                raise StorageError("Failed to complete multipart upload for {}".format(key)) from ex
+
+        self.log.info(
+            "Multipart upload of %r complete, size: %r, took: %.2fs", key, size,
+            time.monotonic() - start_of_multipart_upload
+        )
+
+    def store_file_object(self, key, fd, *, cache_control=None, metadata=None, mimetype=None, upload_progress_fn=None):
+        self.multipart_upload_file_object(
+            cache_control=cache_control,
+            fp=fd,
+            key=key,
+            metadata=metadata,
+            mimetype=mimetype,
+            progress_fn=upload_progress_fn
+        )
+
+    def check_or_create_bucket(self):
+        create_bucket = False
+        try:
+            self.s3_client.head_bucket(Bucket=self.bucket_name)
+        except botocore.exceptions.ClientError as ex:
+            status_code = ex.response.get("ResponseMetadata", {}).get("HTTPStatusCode")
+            if status_code == 301:
+                raise InvalidConfigurationError("Wrong region for bucket {}, check configuration".format(self.bucket_name))
+            elif status_code == 403:
+                self.log.warning("Access denied on bucket check, assuming write permissions")
+            elif status_code == 404:
+                create_bucket = True
+            else:
+                raise
+
+        if create_bucket:
+            self.log.debug("Creating bucket: %r in location: %r", self.bucket_name, self.region)
+            args = {
+                "Bucket": self.bucket_name,
+            }
+            if self.location:
+                args["CreateBucketConfiguration"] = {
+                    "LocationConstraint": self.location,
+                }
+
+            self.s3_client.create_bucket(**args)
+
+    @classmethod
+    def _read_bytes(cls, stream, length):
+        bytes_remaining = length
+        read_results = []
+        while bytes_remaining > 0:
+            data = stream.read(bytes_remaining)
+            if data:
+                read_results.append(data)
+                bytes_remaining -= len(data)
+            else:
+                break
+
+        if not read_results:
+            return None
+        elif len(read_results) == 1:
+            return read_results[0]
+        else:
+            return b"".join(read_results)
diff --git a/rohmu/object_storage/sftp.py b/rohmu/object_storage/sftp.py
new file mode 100644
index 00000000..8bcde4a3
--- /dev/null
+++ b/rohmu/object_storage/sftp.py
@@ -0,0 +1,224 @@
+"""
+rohmu - sftp object store interface
+
+Copyright (c) 2016 Ohmu Ltd
+See LICENSE for details
+"""
+
+import datetime
+import json
+import logging
+import os
+import warnings
+from io import BytesIO, StringIO
+from stat import S_ISDIR
+
+import paramiko
+
+from ..errors import (FileNotFoundFromStorageError, InvalidConfigurationError, StorageError)
+from .base import KEY_TYPE_OBJECT, KEY_TYPE_PREFIX, BaseTransfer, IterKeyItem
+
+
+class SFTPTransfer(BaseTransfer):
+    def __init__(self, server, port, username, password=None, private_key=None, prefix=None):
+        super().__init__(prefix=prefix)
+        self.server = server
+        self.port = port
+        self.username = username
+        self.password = password
+        self.private_key = private_key
+
+        if not password and not private_key:
+            raise InvalidConfigurationError("Password or private key required")
+
+        logging.getLogger("paramiko").setLevel(logging.WARNING)
+
+        # https://github.com/paramiko/paramiko/issues/1386#issuecomment-470847772
+        warnings.filterwarnings(action="ignore", module=".*paramiko.*")
+
+        transport = paramiko.Transport((self.server, self.port))
+
+        if private_key:
+            pkey = paramiko.RSAKey.from_private_key_file(self.private_key)
+            transport.connect(username=self.username, pkey=pkey)
+        else:  # password must be defined due to previous check above
+            transport.connect(username=self.username, password=self.password)
+
+        self.client = paramiko.SFTPClient.from_transport(transport)
+
+        self.log.debug("SFTPTransfer initialized")
+
+    def get_contents_to_file(self, key, filepath_to_store_to, *, progress_callback=None):
+        with open(filepath_to_store_to, "wb") as fh:
+            return self.get_contents_to_fileobj(key, fh, progress_callback=progress_callback)
+
+    def get_contents_to_fileobj(self, key, fileobj_to_store_to, *, progress_callback=None):
+        self._get_contents_to_fileobj(key, fileobj_to_store_to, progress_callback)
+        return self.get_metadata_for_key(key)
+
+    def _get_contents_to_fileobj(self, key, fileobj_to_store_to, progress_callback=None):
+        target_path = self.format_key_for_backend(key.strip("/"))
+        self.log.debug("Get file content: %r", target_path)
+
+        try:
+            # the paramiko progress callback has the same interface as pghoard for downloads
+            return self.client.getfo(remotepath=target_path, fl=fileobj_to_store_to, callback=progress_callback)
+        except FileNotFoundError as ex:
+            raise FileNotFoundFromStorageError(key) from ex
+
+    def get_contents_to_string(self, key):
+        bio = BytesIO()
+        metadata = self.get_contents_to_fileobj(key, bio)
+        return bio.getvalue(), metadata
+
+    def get_file_size(self, key):
+        target_path = self.format_key_for_backend(key.strip("/"))
+        try:
+            return self.client.stat(target_path).st_size
+        except FileNotFoundError as ex:
+            raise FileNotFoundFromStorageError(key) from ex
+
+    def get_metadata_for_key(self, key):
+        bio = BytesIO()
+        self._get_contents_to_fileobj(key + ".metadata", bio)
+        return json.loads(bio.getvalue().decode())
+
+    @staticmethod
+    def _skip_file_name(file_name):
+        return file_name.startswith(".") or file_name.endswith(".metadata") or ".metadata_tmp" in file_name
+
+    def iter_key(self, key, *, with_metadata=True, deep=False, include_key=False):
+        target_path = self.format_key_for_backend(key.strip("/"))
+        self.log.debug("Listing path: %r", target_path)
+
+        try:
+            attrs = self.client.listdir_attr(target_path)
+        except FileNotFoundError:  # if not a directory will throw exception
+            if include_key:
+                file_name = os.path.basename(target_path)
+                if self._skip_file_name(file_name):
+                    return
+
+                try:
+                    attr = self.client.stat(target_path)
+
+                    if with_metadata:
+                        metadata = self.get_metadata_for_key(key)
+                    else:
+                        metadata = None
+
+                    last_modified = datetime.datetime.fromtimestamp(attr.st_mtime, tz=datetime.timezone.utc)
+                    yield IterKeyItem(
+                        type=KEY_TYPE_OBJECT,
+                        value={
+                            "name": key,
+                            "size": attr.st_size,
+                            "last_modified": last_modified,
+                            "metadata": metadata,
+                        }
+                    )
+                    return
+                except FileNotFoundError:
+                    return
+            else:
+                return
+
+        files = set(attr.filename for attr in attrs)
+
+        for attr in attrs:
+            if self._skip_file_name(attr.filename):
+                continue
+
+            file_key = os.path.join(key.strip("/"), attr.filename)
+            if S_ISDIR(attr.st_mode):
+                if deep:
+                    yield from self.iter_key(file_key, with_metadata=with_metadata, deep=True)
+                else:
+                    yield IterKeyItem(type=KEY_TYPE_PREFIX, value=file_key)
+            else:
+                # Don't return files if metadata file is not present; files are written in two phases and
+                # should be considered available only after also metadata has been written
+                if attr.filename + ".metadata" in files:
+                    if with_metadata:
+                        metadata = self.get_metadata_for_key(file_key)
+                    else:
+                        metadata = None
+
+                    last_modified = datetime.datetime.fromtimestamp(attr.st_mtime, tz=datetime.timezone.utc)
+                    yield IterKeyItem(
+                        type=KEY_TYPE_OBJECT,
+                        value={
+                            "name": file_key,
+                            "size": attr.st_size,
+                            "last_modified": last_modified,
+                            "metadata": metadata,
+                        }
+                    )
+
+    # can't support remote copy, only remote rename
+    def copy_file(self, *, source_key, destination_key, metadata=None, **_kwargs):
+        raise NotImplementedError
+
+    def delete_key(self, key):
+        target_path = self.format_key_for_backend(key.strip("/"))
+        self.log.info("Removing path: %r", target_path)
+
+        try:
+            self.client.remove(target_path + ".metadata")
+            self.client.remove(target_path)
+        except FileNotFoundError as ex:
+            raise FileNotFoundFromStorageError(key) from ex
+
+    def store_file_from_memory(self, key, memstring, metadata=None, cache_control=None, mimetype=None):
+        bio = BytesIO(memstring)
+        try:
+            self.store_file_object(key=key, fd=bio, cache_control=cache_control, metadata=metadata, mimetype=mimetype)
+        except OSError as ex:
+            raise StorageError(key) from ex
+
+    def store_file_from_disk(self, key, filepath, metadata=None, multipart=None, cache_control=None, mimetype=None):
+        with open(filepath, "rb") as fh:
+            self.store_file_object(key=key, fd=fh, cache_control=cache_control, metadata=metadata, mimetype=mimetype)
+
+    # pylint: disable=unused-argument
+    def store_file_object(self, key, fd, *, cache_control=None, metadata=None, mimetype=None, upload_progress_fn=None):
+        target_path = self.format_key_for_backend(key.strip("/"))
+
+        self.log.debug("Store path: %r", target_path)
+
+        def wrapper_upload_progress_fn(bytes_written, total_bytes):  # pylint: disable=unused-argument
+            if upload_progress_fn:
+                upload_progress_fn(bytes_written)
+
+        self._mkdir_p(os.path.dirname(target_path))
+        self.client.putfo(fl=fd, remotepath=target_path, callback=wrapper_upload_progress_fn)
+
+        # metadata is saved last, because we ignore data files until the metadata file exists
+        # see iter_key above
+        self._save_metadata(target_path, metadata)
+
+    def _save_metadata(self, target_path, metadata):
+        metadata_path = target_path + ".metadata"
+        self.log.debug("Save metadata: %r", metadata_path)
+
+        sanitised = self.sanitize_metadata(metadata)
+        bio = StringIO(json.dumps(sanitised))
+        self.client.putfo(fl=bio, remotepath=metadata_path)
+
+    # https://stackoverflow.com/questions/14819681/upload-files-using-sftp-in-python-but-create-directories-if-path-doesnt-exist
+    def _mkdir_p(self, remote):
+        dirs_ = []
+        dir_ = remote
+        while len(dir_) > 1:
+            dirs_.append(dir_)
+            dir_, _ = os.path.split(dir_)
+
+        if len(dir_) == 1 and not dir_.startswith("/"):
+            dirs_.append(dir_)  # For a remote path like y/x.txt
+
+        while len(dirs_):
+            dir_ = dirs_.pop()
+            try:
+                self.client.stat(dir_)
+            except OSError:
+                self.client.mkdir(dir_)
diff --git a/rohmu/object_storage/swift.py b/rohmu/object_storage/swift.py
new file mode 100644
index 00000000..f150db4a
--- /dev/null
+++ b/rohmu/object_storage/swift.py
@@ -0,0 +1,351 @@
+"""
+rohmu - openstack swift object store interface
+
+Copyright (c) 2016 Ohmu Ltd
+See LICENSE for details
+"""
+import logging
+import os
+import time
+from contextlib import suppress
+
+from swiftclient import client, exceptions  # pylint: disable=import-error
+
+from ..dates import parse_timestamp
+from ..errors import FileNotFoundFromStorageError
+from .base import KEY_TYPE_OBJECT, KEY_TYPE_PREFIX, BaseTransfer, IterKeyItem
+
+CHUNK_SIZE = 1024 * 1024 * 5  # 5 Mi
+SEGMENT_SIZE = 1024 * 1024 * 1024 * 3  # 3 Gi
+
+
+# Swift client logs excessively at INFO level, outputting things like a curl
+# command line to recreate the request that failed with a simple 404 error.
+# At WARNING level curl commands are not logged, but we get a full ugly
+# traceback for all failures, including 404s.  Monkey-patch them away.
+def swift_exception_logger(err):
+    if not isinstance(err, exceptions.ClientException):
+        return orig_swift_exception_logger(err)
+    if getattr(err, "http_status", None) is None:
+        return orig_swift_exception_logger(err)
+    if err.http_status == 404 and err.msg.startswith("Object GET failed"):
+        client.logger.debug("GET %r FAILED: %r", err.http_path, err.http_status)
+    else:
+        client.logger.error(str(err))
+    return None
+
+
+orig_swift_exception_logger = client.logger.exception
+client.logger.exception = swift_exception_logger
+logging.getLogger("swiftclient").setLevel(logging.WARNING)
+
+
+class SwiftTransfer(BaseTransfer):
+    def __init__(
+        self,
+        *,
+        user,
+        key,
+        container_name,
+        auth_url,
+        auth_version="2.0",
+        tenant_name=None,
+        prefix=None,
+        segment_size=SEGMENT_SIZE,
+        region_name=None,
+        user_id=None,
+        user_domain_id=None,
+        user_domain_name=None,
+        tenant_id=None,
+        project_id=None,
+        project_name=None,
+        project_domain_id=None,
+        project_domain_name=None,
+        service_type=None,
+        endpoint_type=None
+    ):
+        prefix = prefix.lstrip("/") if prefix else ""
+        super().__init__(prefix=prefix)
+        self.container_name = container_name
+
+        if auth_version == "3.0":
+            os_options = {
+                "region_name": region_name,
+                "user_id": user_id,
+                "user_domain_id": user_domain_id,
+                "user_domain_name": user_domain_name,
+                "tenant_id": tenant_id,
+                "project_id": project_id,
+                "project_name": project_name,
+                "project_domain_id": project_domain_id,
+                "project_domain_name": project_domain_name,
+                "service_type": service_type,
+                "endpoint_type": endpoint_type
+            }
+        else:
+            if region_name is not None:
+                os_options = {"region_name": region_name}
+            else:
+                os_options = None
+
+        self.conn = client.Connection(
+            user=user, key=key, authurl=auth_url, tenant_name=tenant_name, auth_version=auth_version, os_options=os_options
+        )
+        self.container = self.get_or_create_container(self.container_name)
+        self.segment_size = segment_size
+        self.log.debug("SwiftTransfer initialized")
+
+    @staticmethod
+    def _headers_to_metadata(headers):
+        return {name[len("x-object-meta-"):]: value for name, value in headers.items() if name.startswith("x-object-meta-")}
+
+    @staticmethod
+    def _metadata_to_headers(metadata):
+        return {"x-object-meta-{}".format(name): str(value) for name, value in metadata.items()}
+
+    def get_metadata_for_key(self, key):
+        key = self.format_key_for_backend(key)
+        return self._metadata_for_key(key)
+
+    def _metadata_for_key(self, key, *, resolve_manifest=False):
+        try:
+            headers = self.conn.head_object(self.container_name, key)
+        except exceptions.ClientException as ex:
+            if ex.http_status == 404:
+                raise FileNotFoundFromStorageError(key)
+            raise
+
+        metadata = self._headers_to_metadata(headers)
+
+        if resolve_manifest and "x-object-manifest" in headers:
+            manifest = headers["x-object-manifest"]
+            seg_container, seg_prefix = manifest.split("/", 1)
+            _, segments = self.conn.get_container(seg_container, prefix=seg_prefix, delimiter="/")
+            segments_size = sum(item["bytes"] for item in segments if "bytes" in item)
+            metadata["_segments_size"] = segments_size
+
+        return metadata
+
+    def iter_key(self, key, *, with_metadata=True, deep=False, include_key=False):
+        path = self.format_key_for_backend(key, trailing_slash=not include_key)
+        self.log.debug("Listing path %r", path)
+        if not deep:
+            kwargs = {"delimiter": "/"}
+        else:
+            kwargs = {}
+        _, results = self.conn.get_container(self.container_name, prefix=path, full_listing=True, **kwargs)
+        for item in results:
+            if "subdir" in item:
+                yield IterKeyItem(type=KEY_TYPE_PREFIX, value=self.format_key_from_backend(item["subdir"]).rstrip("/"))
+            else:
+                if with_metadata:
+                    metadata = self._metadata_for_key(item["name"], resolve_manifest=True)
+                    segments_size = metadata.pop("_segments_size", 0)
+                else:
+                    metadata = None
+                    segments_size = 0
+                last_modified = parse_timestamp(item["last_modified"])
+                yield IterKeyItem(
+                    type=KEY_TYPE_OBJECT,
+                    value={
+                        "name": self.format_key_from_backend(item["name"]),
+                        "size": item["bytes"] + segments_size,
+                        "last_modified": last_modified,
+                        "metadata": metadata,
+                        "hash": item.get("hash"),
+                    },
+                )
+
+    def _delete_object_plain(self, key):
+        try:
+            return self.conn.delete_object(self.container_name, key)
+        except exceptions.ClientException as ex:
+            if ex.http_status == 404:
+                raise FileNotFoundFromStorageError(key)
+            raise
+
+    def _delete_object_segments(self, key, manifest):
+        self._delete_object_plain(key)
+        seg_container, seg_prefix = manifest.split("/", 1)
+        _, segments = self.conn.get_container(seg_container, prefix=seg_prefix, delimiter="/")
+        for item in segments:
+            if "name" in item:
+                with suppress(FileNotFoundFromStorageError):
+                    self._delete_object_plain(item["name"])
+
+    def delete_key(self, key):
+        key = self.format_key_for_backend(key)
+        self.log.debug("Deleting key: %r", key)
+        try:
+            headers = self.conn.head_object(self.container_name, key)
+        except exceptions.ClientException as ex:
+            if ex.http_status == 404:
+                raise FileNotFoundFromStorageError(key)
+            raise
+        if "x-object-manifest" in headers:
+            self._delete_object_segments(key, headers["x-object-manifest"])
+        else:
+            self._delete_object_plain(key)
+
+    def get_contents_to_file(self, key, filepath_to_store_to, *, progress_callback=None):
+        temp_filepath = "{}~".format(filepath_to_store_to)
+        try:
+            with open(temp_filepath, "wb") as fp:
+                metadata = self.get_contents_to_fileobj(key, fp, progress_callback=progress_callback)
+                os.rename(temp_filepath, filepath_to_store_to)
+        finally:
+            with suppress(FileNotFoundError):
+                os.unlink(temp_filepath)
+        return metadata
+
+    def get_contents_to_fileobj(self, key, fileobj_to_store_to, *, progress_callback=None):
+        key = self.format_key_for_backend(key)
+        try:
+            headers, data_gen = self.conn.get_object(self.container_name, key, resp_chunk_size=CHUNK_SIZE)
+        except exceptions.ClientException as ex:
+            if ex.http_status == 404:
+                raise FileNotFoundFromStorageError(key)
+            raise
+
+        content_len = int(headers.get("content-length") or 0)
+        current_pos = 0
+        for chunk in data_gen:
+            fileobj_to_store_to.write(chunk)
+            if progress_callback:
+                if not content_len:
+                    # if content length is not known we'll always say we're half-way there
+                    progress_callback(1, 2)
+                else:
+                    current_pos += len(chunk)
+                    progress_callback(current_pos, content_len)
+
+        return self._headers_to_metadata(headers)
+
+    def get_contents_to_string(self, key):
+        key = self.format_key_for_backend(key)
+        self.log.debug("Starting to fetch the contents of: %r", key)
+        try:
+            headers, data = self.conn.get_object(self.container_name, key)
+        except exceptions.ClientException as ex:
+            if ex.http_status == 404:
+                raise FileNotFoundFromStorageError(key)
+            raise
+
+        metadata = self._headers_to_metadata(headers)
+        return data, metadata
+
+    def get_file_size(self, key):
+        # Not implemented due to lack of environment where to test this. This method is not required by
+        # PGHoard itself, this is only called by external apps that utilize PGHoard's object storage abstraction.
+        raise NotImplementedError
+
+    def store_file_from_memory(self, key, memstring, metadata=None, cache_control=None, mimetype=None):
+        if cache_control is not None:
+            raise NotImplementedError("SwiftTransfer: cache_control support not implemented")
+
+        key = self.format_key_for_backend(key)
+        metadata_to_send = self._metadata_to_headers(self.sanitize_metadata(metadata))
+        self.conn.put_object(
+            self.container_name, key, contents=bytes(memstring), content_type=mimetype, headers=metadata_to_send
+        )
+
+    def store_file_from_disk(self, key, filepath, metadata=None, multipart=None, cache_control=None, mimetype=None):
+        obsz = os.path.getsize(filepath)
+        with open(filepath, "rb") as fp:
+            self._store_file_contents(
+                key,
+                fp,
+                metadata=metadata,
+                multipart=multipart,
+                cache_control=cache_control,
+                mimetype=mimetype,
+                content_length=obsz
+            )
+
+    def get_or_create_container(self, container_name):
+        start_time = time.monotonic()
+        try:
+            self.conn.get_container(container_name, headers={}, limit=1)  # Limit 1 here to not traverse the entire folder
+        except exceptions.ClientException as ex:
+            if ex.http_status == 404:
+                self.conn.put_container(container_name, headers={})
+                self.log.debug(
+                    "Created container: %r successfully, took: %.3fs",
+                    container_name,
+                    time.monotonic() - start_time,
+                )
+                return container_name
+            raise
+        return container_name
+
+    def copy_file(self, *, source_key, destination_key, metadata=None, **_kwargs):
+        source_key = self.format_key_for_backend(source_key)
+        destination_key = "/".join((self.container_name, self.format_key_for_backend(destination_key)))
+        headers = self._metadata_to_headers(self.sanitize_metadata(metadata))
+        if metadata:
+            headers["X-Fresh-Metadata"] = True
+        self.conn.copy_object(self.container_name, source_key, destination=destination_key, headers=headers)
+
+    def store_file_object(self, key, fd, *, cache_control=None, metadata=None, mimetype=None, upload_progress_fn=None):
+        metadata = metadata or {}
+        self._store_file_contents(
+            key,
+            fd,
+            cache_control=cache_control,
+            metadata=metadata,
+            mimetype=mimetype,
+            upload_progress_fn=upload_progress_fn,
+            multipart=True,
+            content_length=metadata.get("Content-Length")
+        )
+
+    def _store_file_contents(
+        self,
+        key,
+        fp,
+        cache_control=None,
+        metadata=None,
+        mimetype=None,
+        upload_progress_fn=None,
+        multipart=None,
+        content_length=None
+    ):
+        if cache_control is not None:
+            raise NotImplementedError("SwiftTransfer: cache_control support not implemented")
+
+        if multipart:
+            # Start by trying to delete the file - if it's a potential multipart file we need to manually
+            # delete it, otherwise old segments won't be cleaned up by anything.  Note that we only issue
+            # deletes with the store_file_from_disk functions, store_file_from_memory is used to upload smaller
+            # chunks.
+            with suppress(FileNotFoundFromStorageError):
+                self.delete_key(key)
+        key = self.format_key_for_backend(key)
+        headers = self._metadata_to_headers(self.sanitize_metadata(metadata))
+        # Fall back to the "one segment" if possible
+        if (not multipart) or (not content_length) or content_length <= self.segment_size:
+            self.log.debug("Uploading %r to %r (%r bytes)", fp, key, content_length)
+            self.conn.put_object(self.container_name, key, contents=fp, content_length=content_length, headers=headers)
+            return
+
+        # Segmented transfer
+        # upload segments of a file like `backup-bucket/site-name/basebackup/2016-03-22_0`
+        # to as `backup-bucket/site-name/basebackup_segments/2016-03-22_0/{:08x}`
+        segment_no = 0
+        segment_path = "{}_segments/{}/".format(os.path.dirname(key), os.path.basename(key))
+        segment_key_format = "{}{{:08x}}".format(segment_path).format
+        remaining = content_length
+        while remaining > 0:
+            this_segment_size = min(self.segment_size, remaining)
+            remaining -= this_segment_size
+            segment_no += 1
+            self.log.debug("Uploading segment %r of %r to %r (%r bytes)", segment_no, fp, key, this_segment_size)
+            segment_key = segment_key_format(segment_no)  # pylint: disable=too-many-format-args
+            self.conn.put_object(
+                self.container_name, segment_key, contents=fp, content_length=this_segment_size, content_type=mimetype
+            )
+            if upload_progress_fn:
+                upload_progress_fn(content_length - remaining)
+        self.log.info("Uploaded %r segments of %r to %r", segment_no, key, segment_path)
+        headers["x-object-manifest"] = "{}/{}".format(self.container_name, segment_path.lstrip("/"))
+        self.conn.put_object(self.container_name, key, contents="", headers=headers, content_length=0)
diff --git a/rohmu/rohmufile.py b/rohmu/rohmufile.py
new file mode 100644
index 00000000..5a0f53e1
--- /dev/null
+++ b/rohmu/rohmufile.py
@@ -0,0 +1,175 @@
+"""
+rohmu - rohmu data transformation interface
+
+Copyright (c) 2016 Ohmu Ltd
+See LICENSE for details
+"""
+
+import time
+from contextlib import suppress
+
+from . import IO_BLOCK_SIZE
+from .compressor import CompressionFile, DecompressionFile, DecompressSink
+from .encryptor import DecryptorFile, DecryptSink, EncryptorFile
+from .errors import InvalidConfigurationError
+from .filewrap import ThrottleSink
+
+
+def _fileobj_name(input_obj):
+    if hasattr(input_obj, "name"):
+        return "open file {!r}".format(getattr(input_obj, "name"))
+
+    return repr(input_obj)
+
+
+def _get_encryption_key_data(metadata, key_lookup):
+    if not metadata or not metadata.get("encryption-key-id"):
+        return None
+
+    key_id = metadata["encryption-key-id"]
+    key_data = None
+    if key_lookup:
+        with suppress(KeyError):
+            key_data = key_lookup(key_id)
+
+    if not key_data:
+        raise InvalidConfigurationError("File is encrypted with key {!r} but key not found".format(key_id))
+    return key_data
+
+
+def file_reader(*, fileobj, metadata=None, key_lookup=None):
+    if not metadata:
+        return fileobj
+
+    key_data = _get_encryption_key_data(metadata, key_lookup)
+    if key_data:
+        fileobj = DecryptorFile(fileobj, key_data)
+
+    comp_alg = metadata.get("compression-algorithm")
+    if comp_alg:
+        fileobj = DecompressionFile(fileobj, comp_alg)
+
+    return fileobj
+
+
+def create_sink_pipeline(*, output, file_size=None, metadata=None, key_lookup=None, throttle_time=0.001):
+    if throttle_time:
+        output = ThrottleSink(output, throttle_time)
+
+    comp_alg = metadata.get("compression-algorithm") if metadata else None
+    if comp_alg:
+        output = DecompressSink(output, comp_alg)
+
+    key_data = _get_encryption_key_data(metadata, key_lookup)
+    if key_data:
+        output = DecryptSink(output, file_size, key_data)
+
+    return output
+
+
+def read_file(*, input_obj, output_obj, metadata, key_lookup, progress_callback=None, log_func=None):
+    start_time = time.monotonic()
+
+    with file_reader(fileobj=input_obj, metadata=metadata, key_lookup=key_lookup) as fp_in:
+        while True:
+            input_data = fp_in.read(IO_BLOCK_SIZE)
+            if not input_data:
+                break
+
+            output_obj.write(input_data)
+            if progress_callback:
+                progress_callback()
+
+    original_size = input_obj.tell()
+    result_size = output_obj.tell()
+
+    if log_func:
+        action = "Decompressed"
+        if metadata.get("encryption-key-id"):
+            action += " and decrypted"
+
+        log_func(
+            "%s %d bytes to %d bytes in %s, took: %.3fs", action, original_size, result_size, _fileobj_name(output_obj),
+            time.monotonic() - start_time
+        )
+
+    return original_size, result_size
+
+
+def file_writer(*, fileobj, compression_algorithm=None, compression_level=0, compression_threads=0, rsa_public_key=None):
+    if rsa_public_key:
+        fileobj = EncryptorFile(fileobj, rsa_public_key)
+
+    if compression_algorithm:
+        fileobj = CompressionFile(fileobj, compression_algorithm, compression_level, compression_threads)
+
+    return fileobj
+
+
+def write_file(
+    *,
+    input_obj,
+    output_obj,
+    progress_callback=None,
+    compression_algorithm=None,
+    compression_level=0,
+    rsa_public_key=None,
+    log_func=None,
+    header_func=None,
+    data_callback=None
+):
+    start_time = time.monotonic()
+
+    original_size = 0
+    with file_writer(
+        fileobj=output_obj,
+        compression_algorithm=compression_algorithm,
+        compression_level=compression_level,
+        rsa_public_key=rsa_public_key
+    ) as fp_out:
+
+        header_block = True
+        while True:
+            input_data = input_obj.read(IO_BLOCK_SIZE)
+            if not input_data:
+                break
+
+            if data_callback:
+                data_callback(input_data)
+
+            if header_block and header_func:
+                header_func(input_data)
+                header_block = False
+
+            fp_out.write(input_data)
+            original_size += len(input_data)
+            if progress_callback:
+                progress_callback()
+
+    result_size = output_obj.tell()
+
+    if log_func:
+        log_compression_result(
+            elapsed=time.monotonic() - start_time,
+            encrypted=bool(rsa_public_key),
+            log_func=log_func,
+            original_size=original_size,
+            result_size=result_size,
+            source_name=_fileobj_name(input_obj),
+        )
+
+    return original_size, result_size
+
+
+def log_compression_result(*, log_func, source_name, original_size, result_size, encrypted, elapsed):
+    if original_size <= result_size:
+        action = "Stored"
+        ratio = ""
+    else:
+        action = "Compressed"
+        ratio = " ({:.0%})".format(result_size / original_size)
+
+    if encrypted:
+        action += " and encrypted"
+
+    log_func("%s %d byte of %s to %d bytes%s, took: %.3fs", action, original_size, source_name, result_size, ratio, elapsed)
diff --git a/rohmu/snappyfile.py b/rohmu/snappyfile.py
new file mode 100644
index 00000000..662426b7
--- /dev/null
+++ b/rohmu/snappyfile.py
@@ -0,0 +1,78 @@
+"""
+rohmu - file-like interface for snappy
+
+Copyright (c) 2016 Ohmu Ltd
+See LICENSE for details
+"""
+
+import io
+
+from . import IO_BLOCK_SIZE
+from .filewrap import FileWrap
+
+try:
+    import snappy
+except ImportError:
+    snappy = None  # type: ignore
+
+
+class SnappyFile(FileWrap):
+    def __init__(self, next_fp, mode):
+        if snappy is None:
+            raise io.UnsupportedOperation("Snappy is not available")
+
+        if mode == "rb":
+            self.decr = snappy.StreamDecompressor()
+            self.encr = None
+        elif mode == "wb":
+            self.decr = None
+            self.encr = snappy.StreamCompressor()
+        else:
+            raise io.UnsupportedOperation("unsupported mode for SnappyFile")
+
+        super().__init__(next_fp)
+        self.decr_done = False
+
+    def close(self):
+        if self.closed:
+            return
+        if self.encr:
+            data = self.encr.flush() or b""
+            if data:
+                self.next_fp.write(data)
+            self.next_fp.flush()
+        super().close()
+
+    def write(self, data):
+        self._check_not_closed()
+        if self.encr is None:
+            raise io.UnsupportedOperation("file not open for writing")
+        compressed_data = self.encr.compress(data)
+        self.next_fp.write(compressed_data)
+        self.offset += len(data)
+        return len(data)
+
+    def writable(self):
+        return self.encr is not None
+
+    def read(self, size=-1):  # pylint: disable=unused-argument
+        # NOTE: size arg is ignored, random size output is returned
+        self._check_not_closed()
+        if self.decr is None:
+            raise io.UnsupportedOperation("file not open for reading")
+        while not self.decr_done:
+            compressed = self.next_fp.read(IO_BLOCK_SIZE)
+            if not compressed:
+                self.decr_done = True
+                output = self.decr.flush()
+            else:
+                output = self.decr.decompress(compressed)
+
+            if output:
+                self.offset += len(output)
+                return output
+
+        return b""
+
+    def readable(self):
+        return self.decr is not None
diff --git a/rohmu/version.py b/rohmu/version.py
new file mode 100644
index 00000000..3277f64c
--- /dev/null
+++ b/rohmu/version.py
@@ -0,0 +1,1 @@
+VERSION = "1.0.0"
diff --git a/rohmu/zstdfile.py b/rohmu/zstdfile.py
new file mode 100644
index 00000000..cce09392
--- /dev/null
+++ b/rohmu/zstdfile.py
@@ -0,0 +1,86 @@
+"""
+rohmu - file-like interface for zstd
+
+Copyright (c) 2016 Ohmu Ltd
+See LICENSE for details
+"""
+
+import io
+
+from . import IO_BLOCK_SIZE
+from .filewrap import FileWrap
+
+try:
+    import zstandard as zstd
+except ImportError:
+    zstd = None  # type: ignore
+
+
+class _ZstdFileWriter(FileWrap):
+    def __init__(self, next_fp, level, threads=0):
+        self._zstd = zstd.ZstdCompressor(level=level, threads=threads).compressobj()
+        super().__init__(next_fp)
+
+    def close(self):
+        if self.closed:
+            return
+        data = self._zstd.flush() or b""
+        if data:
+            self.next_fp.write(data)
+        self.next_fp.flush()
+        super().close()
+
+    def write(self, data):
+        self._check_not_closed()
+        compressed_data = self._zstd.compress(data)
+        self.next_fp.write(compressed_data)
+        self.offset += len(data)
+        return len(data)
+
+    def writable(self):
+        return True
+
+
+class _ZtsdFileReader(FileWrap):
+    def __init__(self, next_fp):
+        self._zstd = zstd.ZstdDecompressor().decompressobj()
+        super().__init__(next_fp)
+        self._done = False
+
+    def close(self):
+        if self.closed:
+            return
+        super().close()
+
+    def read(self, size=-1):  # pylint: disable=unused-argument
+        # NOTE: size arg is ignored, random size output is returned
+        self._check_not_closed()
+        while not self._done:
+            compressed = self.next_fp.read(IO_BLOCK_SIZE)
+            if not compressed:
+                self._done = True
+                output = self._zstd.flush() or b""
+            else:
+                output = self._zstd.decompress(compressed)
+
+            if output:
+                self.offset += len(output)
+                return output
+
+        return b""
+
+    def readable(self):
+        return True
+
+
+def open(fp, mode, level=0, threads=0):  # pylint: disable=redefined-builtin
+    if zstd is None:
+        raise io.UnsupportedOperation("zstd is not available")
+
+    if mode == "wb":
+        return _ZstdFileWriter(fp, level, threads)
+
+    if mode == "rb":
+        return _ZtsdFileReader(fp)
+
+    raise io.UnsupportedOperation("unsupported mode for zstd")
diff --git a/setup.cfg b/setup.cfg
new file mode 100644
index 00000000..7c3f5bd5
--- /dev/null
+++ b/setup.cfg
@@ -0,0 +1,43 @@
+[metadata]
+name = rohmu
+version = attr: rohmu.version.VERSION
+description = "Rohmu is a python library providing an interface to various cloud storage providers."
+long_description = file: README.md
+license = Apache 2.0
+classifiers =
+    Development Status :: 5 - Production/Stable
+    Intended Audience :: Developers,
+    Intended Audience :: Information Technology,
+    Intended Audience :: System Administrators,
+    License :: OSI Approved :: Apache Software License,
+    Programming Language :: Python :: 3.6,
+    Programming Language :: Python :: 3.7,
+    Programming Language :: Python :: 3.8,
+    Programming Language :: Python :: 3.9,
+    Programming Language :: Python :: 3.10,
+    Topic :: Database :: Database Engines/Servers,
+    Topic :: Software Development :: Libraries :: Python Modules
+url = https://github.com/aiven/rohmu/
+platforms =
+    POSIX
+    MacOS
+[options]
+package_dir =
+    =.
+packages = find:
+install_requires =
+    azure-storage-blob == 2.1.0
+    botocore
+    cryptography
+    google-api-python-client
+    httplib2
+    oauth2client
+    paramiko
+    pydantic
+    python-dateutil
+    python-snappy
+    requests
+    zstandard
+[options.packages.find]
+where = .
+include = rohmu
diff --git a/setup.py b/setup.py
new file mode 100644
index 00000000..dcfd6b58
--- /dev/null
+++ b/setup.py
@@ -0,0 +1,8 @@
+#!/usr/bin/env python3
+
+# This file enables editable installs with `pip install --editable .`
+# https://setuptools.readthedocs.io/en/latest/setuptools.html#setup-cfg-only-projects
+
+import setuptools
+
+setuptools.setup()


Test patch:
diff --git a/test/conftest.py b/test/conftest.py
new file mode 100644
index 00000000..0a1b7afe
--- /dev/null
+++ b/test/conftest.py
@@ -0,0 +1,7 @@
+import pytest
+from rohmu.object_storage.local import LocalTransfer
+
+@pytest.fixture(name="local_transfer")
+def local_transfer(tmp_path):
+    return LocalTransfer(tmp_path / "local_storage")
+
diff --git a/test/test_storage.py b/test/test_storage.py
new file mode 100644
index 00000000..48746ffe
--- /dev/null
+++ b/test/test_storage.py
@@ -0,0 +1,43 @@
+import pytest
+from conftest import local_transfer
+from rohmu import errors
+from io import BytesIO
+
+@pytest.mark.parametrize("transfer",
+                         ["local_transfer"])
+def test_nonexistent(transfer, request):
+    transfer = request.getfixturevalue(transfer)
+    with pytest.raises(errors.FileNotFoundFromStorageError):
+        transfer.get_metadata_for_key("NONEXISTENT")
+    with pytest.raises(errors.FileNotFoundFromStorageError):
+        transfer.delete_key("NONEXISTENT")
+    with pytest.raises(errors.FileNotFoundFromStorageError):
+        transfer.get_contents_to_file("NONEXISTENT", "nonexistent/a")
+    with pytest.raises(errors.FileNotFoundFromStorageError):
+        transfer.get_contents_to_fileobj("NONEXISTENT", BytesIO())
+    with pytest.raises(errors.FileNotFoundFromStorageError):
+        transfer.get_contents_to_string("NONEXISTENT")
+    assert transfer.list_path("") == []
+    assert transfer.list_path("NONEXISTENT") == []
+
+@pytest.mark.parametrize("transfer",
+                         ["local_transfer"])
+def test_basic_upload(transfer, tmp_path, request):
+    scratch = tmp_path / "scratch"
+    scratch.mkdir()
+    transfer = request.getfixturevalue(transfer)
+    transfer.store_file_from_memory("x1", b"dummy", {"k": "v"})
+    assert transfer.get_contents_to_string("x1") == (b"dummy", {"k": "v"})
+    # Same thing, but with a key looking like a directory
+    transfer.store_file_from_memory("NONEXISTENT-DIR/x1", b"dummy", None)
+    transfer.get_contents_to_string("x1") == (b"dummy", None)
+
+    # Same thing, but from disk now
+    dummy_file= scratch / "a"
+    with open(dummy_file, "wb") as fp:
+        fp.write(b"dummy")
+    transfer.store_file_from_disk("test1/x1", dummy_file, None)
+    out = BytesIO()
+
+    assert transfer.get_contents_to_fileobj("test1/x1", out) == {}
+    assert out.getvalue() == b"dummy"