API Reference

This API reference describes each component of the AQIS staging library.

`cli.py`

Overview

cli.py provides the command-line entry point for the streaming staging workflow.

it loads workflow configuration from an INI file
optionally retrieves secrets from Infisical
creates source and destination storage adaptors
builds a source file manifest
and streams files directly from the source backend to the destination backend.

The workflow supports local filesystem, S3, iRODS, and SCP backends. Transfers are executed in parallel using worker processes with each file streamed in chunks. The workflow validates transfers using file size checks but also provides the option to perform full checksum validation.

Module Constants

ADAPTOR_MAP

ADAPTOR_MAP = {
    "FS": FSAdaptor,
    "S3": S3Adaptor,
    "IRODS": IRODSAdaptor,
    "SCP": SCPAdaptor,
}

Maps configured storage backend names to their corresponding adaptor classes.

Supported values:

Type	Adaptor
`FS`	`FSAdaptor`
`S3`	`S3Adaptor`
`IRODS`	`IRODSAdaptor`
`SCP`	`SCPAdaptor`

Used to instantiate source and destination adaptors based on SOURCE_TYPE and DEST_TYPE.

AdaptorClass

AdaptorClass = TypeVar("AdaptorClass", bound=StorageAdaptor)

Type variable representing a concrete storage adaptor class derived from StorageAdaptor.

ManifestType

ManifestType = tuple[str, int]

Represents a single manifest entry.

The tuple contains:

Index	Type	Description
`0`	`str`	Relative file path
`1`	`int`	Expected file size in bytes

Example:

("example.json", 2048)

_worker_runner

_worker_runner: "StreamingFileRunner | None" = None

Global worker-local StreamingFileRunner instance.

Each worker process initialises its own runner using init_worker().

_stream_chunk_size

_stream_chunk_size = 64 * 1024 * 1024

Global stream chunk size used by worker processes.

Default value is 64 MiB.

This can be configured through:

STREAM_CHUNK_SIZE_MB=64

Functions

init_worker()

def init_worker(
    source_path: str,
    upload_dest_base: str,
    source_adaptor_cls: Type[AdaptorClass],
    dest_adaptor_cls: Type[AdaptorClass],
    stream_chunk_size: int,
    checksum_strategy: str,
    checksum_algorithm: str,
):

Initialises a worker process for streaming transfers.

This function is passed to ProcessPoolExecutor as the worker initializer. It creates a worker-local StreamingFileRunner and stores it in the global _worker_runner.

Name	Type	Description
`source_path`	`str`	Base source path used to resolve manifest entries
`upload_dest_base`	`str`	Base destination path used for uploads
`source_adaptor_cls`	`Type[StorageAdaptor]`	Source adaptor class
`dest_adaptor_cls`	`Type[StorageAdaptor]`	Destination adaptor class
`stream_chunk_size`	`int`	Chunk size in bytes
`checksum_strategy`	`str`	Checksum strategy, usually `SIZE` or `FULL`
`checksum_algorithm`	`str`	Hashing algorithm, for example `md5`

Behaviour

Creates adaptor instances using the configured prefixes:

source_adaptor_cls("SRC")
dest_adaptor_cls("DST")

The prefixes are used by the adaptors to read environment variables such as:

SRC_S3_ENDPOINT_URL
DST_IRODS_HOST

process_item()

def process_item(item: ManifestType, free_indices: Queue):

Processes a single manifest item in a worker process.

This function obtains a progress-bar position from the shared queue, streams the file using the worker-local StreamingFileRunner, and returns the position to the queue when complete.

Name	Type	Description
`item`	`ManifestType`	Manifest entry containing relative path and expected size
`free_indices`	`Queue`	Shared queue of available tqdm progress-bar positions

Returns

Returns the manifest item after successful transfer.

Raises

May raise any exception raised by:

_worker_runner.stream_single_file()

Typical failures include:

Error	Cause
`RuntimeError`	Size mismatch
`RuntimeError`	Checksum mismatch
Storage adaptor exception	Read/write failure
Network exception	Backend connection failure

dest_path_check()

def dest_path_check(source_path: str, dest_path: str, single_file: bool) -> str:

Determines the final base destination path.

For directory transfers, if the source folder name does not match the destination folder name, the source folder name is appended to the destination path.

Name	Type	Description
`source_path`	`str`	Source file or directory path
`dest_path`	`str`	Configured destination path
`single_file`	`bool`	Whether the transfer contains a single source file

Returns

The resolved destination base path.

Example

dest_path_check(
    source_path="/data/my_folder",
    dest_path="/tempZone/home/rods",
    single_file=False,
)

Returns:

/tempZone/home/rods/my_folder

If single_file=True, the original dest_path is returned unchanged.

ini_has_section()

def ini_has_section(path: str, section: str) -> bool:

Name	Type	Description
`path`	`str`	Path to the INI file
`section`	`str`	Section name to check

Returns

True if the section exists, otherwise False.

Example

ini_has_section("config.ini", "creds")

parse_args()

def parse_args() -> argparse.Namespace:

Parses command-line arguments for the staging workflow.

Supported Arguments

Argument	Required	Default	Description
`--config`	No	`config.ini`	Main workflow INI file
`--infisical-src-config`	No	`None`	Optional Infisical config for source secrets
`--infisical-dst-config`	No	`None`	Optional Infisical config for destination secrets

Returns

An argparse.Namespace containing the parsed command-line arguments.

Example

python -m staging_workflow.cli \
  --config configs/workflows/s3_to_irods.ini \
  --infisical-src-config configs/infisical/src.ini \
  --infisical-dst-config configs/infisical/dst.ini

load_configuration()

def load_configuration(args: argparse.Namespace) -> None:

Loads workflow configuration into environment variables.

The function always loads the [env] section from the main configuration file. It then either loads secrets from Infisical or falls back to the [creds] section of the main configuration file.

Name	Type	Description
`args`	`argparse.Namespace`	Parsed command-line arguments

Behaviour

The configuration loading order is:

Load [env] from args.config.
If Infisical source or destination configs are provided:
Load secrets from Infisical.
Do not load [creds] from the main config.
If no Infisical config is provided:
Load [creds] from the main config if it exists.

Example

args = parse_args()
load_configuration(args)

main()

def main():

Main entry point for the streaming staging workflow.

This function coordinates the complete transfer lifecycle.

Workflow Steps

Parse command-line arguments.
Load configuration and secrets.
Read required environment variables.
Validate source and destination backend types.
Create source adaptor for manifest generation.
Run source compliance checks.
Determine destination path layout.
Create a pool of worker processes.
Stream files in parallel.
Validate file sizes and optional checksums.
Report bandwidth and transfer result.

Required Environment Variables

Variable	Description
`SOURCE_TYPE`	Source backend type: `FS`, `S3`, `IRODS`, or `SCP`
`DEST_TYPE`	Destination backend type: `FS`, `S3`, `IRODS`, or `SCP`
`SOURCE_PATH`	Source path to transfer
`DEST_PATH`	Destination path

Optional Environment Variables

Variable	Default	Description
`MAX_FILE_SIZE_MB`	Unlimited	Maximum allowed file size in MiB
`CONCURRENCY`	`10`	Number of parallel worker processes
`STREAM_CHUNK_SIZE_MB`	`64`	Streaming chunk size in MiB
`CHECKSUM_STRATEGY`	`SIZE`	Validation strategy
`CHECKSUM_ALGORITHM`	`md5`	Hashing algorithm for full checksums

Checksum Strategies

Strategy	Description
`SIZE`	Validate expected byte count and destination size
`FULL`	Hash source stream and compare with destination checksum

Note

The manifest compliance check currently uses:

IntegrityManager("SIZE", max_file_size_mb)

This means the manifest is built using size-based validation and file size limits. Full checksum validation is handled later during streaming by StreamingFileRunner.

Classes

`ProgressReader`

class ProgressReader:

File-like wrapper used to track upload progress and optionally update a checksum while a stream is being read.

This is mainly used for destination adaptors such as S3Adaptor, where uploads may consume a file-like object directly.

Constructor

def __init__(self, raw, pbar, hasher=None):

Name	Type	Description
`raw`	file-like object	Underlying readable stream
`pbar`	`tqdm`	Progress bar updated as bytes are read
`hasher`	hash object or `None`	Optional hashlib object used for checksum updates

Attributes

Attribute	Description
`raw`	Wrapped input stream
`pbar`	Progress bar instance
`hasher`	Optional checksum hasher
`bytes_read`	Total number of bytes read

Methods

read

def read(self, size=-1):

Reads bytes from the wrapped stream.

Updates:

progress bar
byte counter
checksum hasher, if enabled

Returns the chunk read from the underlying stream.

close

def close(self):

Closes the wrapped stream.

__enter__()

def __enter__(self):

Context manager entry method.

__exit__()

def __exit__(self, exc_type, exc, tb):

Context manager exit method. Closes the wrapped stream.

`StreamingFileRunner`

@dataclasses.dataclass
class StreamingFileRunner:

Handles streaming transfer of individual files from a source adaptor to a destination adaptor.

Each worker process owns one StreamingFileRunner instance.

Field	Type	Description
`source_path`	`str`	Base source path
`upload_dest_base`	`str`	Base destination path
`source_adaptor`	`StorageAdaptor`	Source storage adaptor
`dest_adaptor`	`StorageAdaptor`	Destination storage adaptor
`checksum_strategy`	`str`	Validation strategy, default `SIZE`
`checksum_algorithm`	`str`	Hashing algorithm, default `md5`

StreamingFileRunner._make_hasher()

def _make_hasher(self):

Creates a checksum hasher when full checksum validation is enabled.

Returns

Condition	Return Value
`checksum_strategy != "FULL"`	`None`
`checksum_strategy == "FULL"`	`hashlib` hasher instance

Example

self.checksum_strategy = "FULL"
self.checksum_algorithm = "md5"

hasher = self._make_hasher()

StreamingFileRunner._normalise_checksum()

def _normalise_checksum(self, value: str | None) -> str | None:

Normalises checksum strings before comparison.

Parameters

Name	Type	Description
`value`	`str \| None`	Raw checksum value

Returns

A cleaned checksum string, or None.

Behaviour

The function:

converts the value to string
strips whitespace
converts to lowercase
removes surrounding quotes
removes algorithm prefixes such as md5: or sha2:

Example

_normalise_checksum("md5:ABC123")

Returns:

abc123

StreamingFileRunner._checksum_by_reading()

def _checksum_by_reading(self, adaptor: StorageAdaptor, path: str) -> str:

Calculates a checksum by reading a file from a storage adaptor.

This is used when adaptor-native checksums are unavailable or when the native checksum format does not match the configured checksum algorithm.

Parameters

Name	Type	Description
`adaptor`	`StorageAdaptor`	Storage adaptor to read from
`path`	`str`	File path to checksum

Returns

Checksum digest as a hexadecimal string.

Example

checksum = self._checksum_by_reading(self.dest_adaptor, "/target/file.json")

StreamingFileRunner._get_destination_checksum()

def _get_destination_checksum(self, dest_file_path: str) -> str:

Returns the destination checksum using the same algorithm as the source checksum.

Parameters

Name	Type	Description
`dest_file_path`	`str`	Destination file path

Returns

Destination checksum string.

Important Behaviour

For md5, the destination file is read back and hashed locally:

if self.checksum_algorithm.lower() == "md5":
    return self._checksum_by_reading(self.dest_adaptor, dest_file_path)

This avoids comparing MD5 against backend-native checksum formats such as iRODS SHA-256/base64 checksums.

For non-MD5 algorithms, the method may attempt to use adaptor-native checksums through:

self.dest_adaptor.get_remote_checksum(dest_file_path)

If that is unavailable, it falls back to reading the destination and hashing it locally.

Raises

Error	Cause
`RuntimeError`	Destination cannot provide or calculate a checksum

StreamingFileRunner.stream_single_file()

def stream_single_file(self, source_file_str: ManifestType, position: int = 1):

Streams a single file from the source adaptor to the destination adaptor.

Parameters

Name	Type	Description
`source_file_str`	`ManifestType`	Manifest entry containing relative path and expected file size
`position`	`int`	tqdm progress-bar position

Returns

The original manifest item after successful transfer.

Behaviour

The method:

Resolves the source file path.
Resolves the destination file path.
Optionally prepares the destination for streaming.
Opens the source file for reading.
Streams the file to the destination.
Updates a progress bar.
Tracks the number of bytes written.
Optionally calculates source checksum while streaming.
Validates byte count.
Validates destination size if supported.
Optionally validates destination checksum.

Source Path Resolution

source_file_path = Path(self.source_path) / source_file

Destination Path Resolution

dest_file_path = Path(self.upload_dest_base) / source_file

Destination Preparation

If the destination adaptor supports:

prepare_for_stream_write()

then the method calls:

self.dest_adaptor.prepare_for_stream_write(
    str(dest_file_path),
    expected_size=expected_size,
)

if this returns True, the file is skipped because a complete destination file already exists.

S3 Upload Behaviour

If the destination adaptor is S3Adaptor, the method uses:

self.dest_adaptor.upload_stream(
    ProgressReader(src, pbar, source_hasher),
    str(dest_file_path),

This allows the S3 adaptor to consume a file-like stream directly.

Generic Upload Behaviour

For non-S3 destinations, the method opens a destination write stream:

with self.dest_adaptor.open_write(str(dest_file_path), overwrite=True) as dst:

Then copies chunks from source to destination:

chunk = src.read(_stream_chunk_size)
dst.write(chunk)

Size Validation

After streaming, the method checks that the number of bytes written matches the expected manifest size:

if bytes_written != expected_size:
    raise RuntimeError(...)

If the destination adaptor supports get_size(), the destination size is also checked.

Checksum Validation

If:

checksum_strategy == "FULL"

the method compares:

source_checksum

against:

dest_checksum

A mismatch raises a RuntimeError.

Raises

Error	Cause
`RuntimeError`	Streamed byte count does not match expected size
`RuntimeError`	Destination size does not match expected size
`RuntimeError`	Source and destination checksums do not match
Backend-specific exception	Source read or destination write failed

Configuration Reference

Main Workflow INI

Example:

[env]
SOURCE_TYPE=S3
DEST_TYPE=IRODS
SOURCE_PATH=/path/to/example_data
DEST_PATH=/tempZone/home/rods/example_data

MAX_FILE_SIZE_MB=5000
CONCURRENCY=10
STREAM_CHUNK_SIZE_MB=64

CHECKSUM_STRATEGY=FULL
CHECKSUM_ALGORITHM=md5

[creds]
SRC_S3_ENDPOINT_URL=https://example-minio:9000
SRC_S3_ACCESS_KEY_ID=...
SRC_S3_SECRET_ACCESS_KEY=...
SRC_S3_REGION=us-east-1
SRC_S3_BUCKET=wp6-health

DST_IRODS_HOST=example-irods-host
DST_IRODS_PORT=1247
DST_IRODS_USER=rods
DST_IRODS_PASSWORD=...
DST_IRODS_ZONE=tempZone

`[env]`

The [env] section defines workflow behaviour.

Key	Required	Description
`SOURCE_TYPE`	Yes	Source backend type
`DEST_TYPE`	Yes	Destination backend type
`SOURCE_PATH`	Yes	Source file or directory
`DEST_PATH`	Yes	Destination file or directory
`MAX_FILE_SIZE_MB`	No	Maximum file size limit
`CONCURRENCY`	No	Number of parallel workers
`STREAM_CHUNK_SIZE_MB`	No	Stream chunk size in MiB
`CHECKSUM_STRATEGY`	No	`SIZE` or `FULL`
`CHECKSUM_ALGORITHM`	No	Hash algorithm, for example `md5`

`[creds]`

The [creds] section stores backend credentials.

It is only loaded when no Infisical configuration is provided.

If either of these arguments is used:

--infisical-src-config
--infisical-dst-config

then [creds] is skipped.

Command-Line Usage

Basic Usage

python -m staging_workflow.cli --config config.ini

With Infisical Secrets

python -m staging_workflow.cli \
  --config configs/workflows/s3_to_irods.ini \
  --infisical-src-config configs/infisical/source.ini \
  --infisical-dst-config configs/infisical/destination.ini

Execution Model

The workflow uses a process pool:

ProcessPoolExecutor(max_workers=concurrency_value)

Each worker process receives:

source path
destination path
source adaptor class
destination adaptor class
stream chunk size
checksum strategy
checksum algorithm

Files are submitted as individual tasks:

pool.submit(process_item, item, free_slots)

Each task streams one file.

Validation Behaviour

Size Validation

Always performed.

The workflow checks:

bytes read/written during streaming
destination size, if the adaptor supports get_size()

Full Checksum Validation

Enabled with:

CHECKSUM_STRATEGY=FULL

The workflow:

Creates a hasher using hashlib.
Updates the hasher while reading the source stream.
Calculates or retrieves the destination checksum.
Compares source and destination checksums.

Recommended setting:

CHECKSUM_STRATEGY=FULL
CHECKSUM_ALGORITHM=md5

For MD5, the destination is read back and hashed locally to avoid comparing against backend-native checksum formats such as iRODS SHA-256/base64.

Error Handling

Worker failures are captured per file.

If a worker task fails:

logger.exception("Worker task failed")
failed_items.append(futures[completed_future])

The workflow continues processing remaining files.

At the end:

Condition	Result
No failed files	Logs success
One or more failed files	Logs failure and lists failed manifest entries

Backend Requirements

Each adaptor should implement the methods needed for the workflow path being used.

Required for Source Adaptors

Method	Purpose
`open_read(path)`	Open source file as readable stream
Manifest/listing support	Used by `IntegrityManager.check_compliance()`
Required for Destination Adaptors

Required for Destination Adaptors

Method	Purpose
`open_write(path, overwrite=True)`	Open destination stream for writing
`upload_stream(stream, path)`	Used for S3-style streaming uploads
`get_size(path)`	Optional destination size validation
`prepare_for_stream_write(path, expected_size)`	Optional skip/resume preparation
`get_remote_checksum(path)`	Optional backend-native checksum retrieval
`open_read(path)`	Required when destination checksum is calculated by reading back

Example Transfer Flow

For an S3 to iRODS transfer:

SOURCE_TYPE=S3
DEST_TYPE=IRODS
SOURCE_PATH=/path/to/example
DEST_PATH=/tempZone/home/rods
CHECKSUM_STRATEGY=FULL
CHECKSUM_ALGORITHM=md5
CONCURRENCY=4
STREAM_CHUNK_SIZE_MB=64

The workflow will:

Build a manifest from the S3 source.
Resolve the destination path.
Start four worker processes.
Stream each S3 object to iRODS.
Track per-file progress.
Verify byte counts.
Read the iRODS object back and calculate MD5.
Compare source MD5 with destination MD5.
Report failed files, if any.

Adaptors

`base.py`

Overview

base.py defines the common adaptor interface used by the staging workflow. It provides the abstract StorageAdaptor base class that all storage backend adaptors must implement, such as filesystem, S3, iRODS, and SCP adaptors.

The module also defines TQDMHolder, a small helper class used to manage tqdm progress bars for transfer operations.

The current version is designed primarily around streaming transfers, where files are read from a source backend and written directly to a destination backend without first downloading the full file to a local temporary directory.

Imports

import dataclasses
import hashlib
from abc import ABC, abstractmethod
from typing import BinaryIO, Generator, Tuple

from tqdm import tqdm

Import	Purpose
`dataclasses`	Used to define `TQDMHolder` as a dataclass
`hashlib`	Previously used for local checksum calculation in the legacy API
`ABC`, `abstractmethod`	Used to define the abstract adaptor interface
`BinaryIO`	Type hint for readable/writable binary streams
`Generator`	Type hint for object listing generators
`Tuple`	Type hint for manifest entries
`tqdm`	Progress bar support

Class: `StorageAdaptor`

class StorageAdaptor(ABC):

Abstract base class for all storage backend adaptors.

Concrete adaptors such as FSAdaptor, S3Adaptor, IRODSAdaptor, and SCPAdaptor should inherit from this class and implement the required methods.

The adaptor interface is designed around three main operations:

Listing source objects. Opening source objects as readable streams. Writing or uploading streams to a destination backend.

Constructor

def __init__(self, prefix):

Initialises the adaptor with an environment-variable prefix.

Parameters

Name	Type	Description
`prefix`	`str`	Prefix used to identify source or destination environment variables

Typical values are:

SRC
DST

For example, an S3 adaptor may use this prefix to read variables such as:

SRC_S3_ENDPOINT_URL
SRC_S3_BUCKET
DST_S3_ENDPOINT_URL
DST_S3_BUCKET

Example

source_adaptor = S3Adaptor("SRC")
dest_adaptor = IRODSAdaptor("DST")

Attribute: prefix

self.prefix = prefix

Stores the adaptor prefix.

This is used by concrete adaptors to load the correct environment variables for source or destination configuration.

Abstract Methods

list_objects()

@abstractmethod
def list_objects(self, path: str) -> Generator[Tuple[str, int], None, None]:

Lists files or objects under a given source path.

This method is used during manifest generation. Each returned item represents one transferable file.

Parameters

Name	Type	Description
`path`	`str`	Source directory, prefix, collection, or file path

Yields

A generator of tuples:

(relative_path, size_in_bytes)

Tuple Item	Type	Description
`relative_path`	`str`	File path relative to the source base path
`size_in_bytes`	`int`	File size in bytes

Example Return Values

yield ("file.json", 2048)
yield ("nested/data/file.csv", 1048576)

Expected Behaviour

Implementations should recursively list files where appropriate.

For example:

Backend	Expected Listing Behaviour
Filesystem	Walk local directories
S3	List objects under a bucket prefix
iRODS	List data objects under a collection
SCP	List files on a remote host

open_read()

@abstractmethod
def open_read(self, source_path: str) -> BinaryIO:

Opens a source file or object as a readable binary stream.

This method is used by the streaming workflow to read source data in chunks.

Name	Type	Description
`source_path`	`str`	Path to the source file or object

Returns

A readable binary stream.

The returned object should support:

read(size)
close()

and ideally context manager usage:

with adaptor.open_read(path) as src:
    chunk = src.read(8192)

Example

with source_adaptor.open_read("/data/file.csv") as src:
    data = src.read(1024)

get_size()

@abstractmethod
def get_size(self, path: str) -> int:

Returns the size of a file or object in bytes.

This is used for compliance checks and destination validation.

Parameters

Name	Type	Description
`path`	`str`	File, object, or collection path

Returns

File size in bytes.

Example

size = adaptor.get_size("/data/file.csv")

Expected Return

Optional Streaming Methods

open_write()

def open_write(self, dest_path: str, overwrite: bool = True) -> BinaryIO:

Opens a destination file or object as a writable binary stream.

Backends that support normal writable file handles should override this method.

For example:

Backend	Suitable for `open_write()`
Local filesystem	Yes
SCP/SFTP	Yes
iRODS	Yes, depending on implementation
S3	Usually no

Parameters

Name	Type	Default	Description
`dest_path`	`str`	Required	Destination file path
`overwrite`	`bool`	`True`	Whether to overwrite an existing destination file

Returns

A writable binary stream.

The returned object should support:

write(bytes)
close()

and ideally context manager usage:

with adaptor.open_write(path, overwrite=True) as dst:
    dst.write(chunk)

Default Behaviour

The base class raises:

NotImplementedError

Raises

NotImplementedError

if the concrete adaptor does not support writable streams.

upload_stream()

def upload_stream(self, stream: BinaryIO, dest_path: str, extra_args=None) -> None:

Uploads a readable stream to the destination backend.

This method is mainly intended for object stores such as S3, where a normal writable file handle is not usually available.

Parameters

Name	Type	Default	Description
`stream`	`BinaryIO`	Required	Readable binary stream to upload
`dest_path`	`str`	Required	Destination object path
`extra_args`	`dict \\| None`	`None`	Optional backend-specific upload arguments

Returns

None

Example

with source_adaptor.open_read("/source/file.csv") as src:
    s3_adaptor.upload_stream(src, "/target/file.csv")

Default Behaviour

The base class raises:

NotImplementedError

Raises

NotImplementedError

if the concrete adaptor does not support stream uploads.

prepare_for_stream_write()

def prepare_for_stream_write(
    self,
    dest_path: str,
    expected_size: int | None = None,
) -> bool:

Optional helper used before streaming a file to the destination.

This can be overridden by adaptors that are able to check whether a destination object already exists and whether it is complete.

Parameters

Name	Type	Default	Description
`dest_path`	`str`	Required	Destination file or object path
`expected_size`	`int \\| None`	`None`	Expected file size in bytes

Returns

Return Value	Meaning
`True`	Destination already exists and should be skipped
`False`	Transfer should proceed

Default Behaviour

return False

The base implementation always tells the workflow to continue with the transfer.

Example Use Case

An adaptor may implement:

def prepare_for_stream_write(self, dest_path, expected_size=None):
    if self.exists(dest_path):
        if expected_size is not None and self.get_size(dest_path) == expected_size:
            return True

    return False

This allows the workflow to skip files that already exist with the correct size.

checksum_remote()

def checksum_remote(self, path: str):

Optional backend-native checksum helper.

This method is intended to delegate to a backend-specific checksum implementation.

Parameters

Name	Type	Description
`path`	`str`	Path to the remote or local file/object

Returns

A backend-native checksum value.

The returned format may depend on the backend.

Examples:

Backend	Possible Checksum Format
Filesystem	MD5 hex
S3	ETag or metadata checksum
iRODS	SHA-256/base64
SCP	Locally calculated checksum after reading

Class: `TQDMHolder`

@dataclasses.dataclass
class TQDMHolder:

Helper class for managing a tqdm progress bar during transfer operations.

This class is mainly useful for adaptor-specific callbacks, such as S3 or SCP progress callbacks.

Fields

filename: str
pbar: tqdm = None
bytes_to_mb = 1e6
amount_transferred: int = 0

Field	Type	Default	Description
`filename`	`str`	Required	File name displayed in the progress bar
`pbar`	`tqdm \\| None`	`None`	Progress bar instance
`bytes_to_mb`	`float`	`1e6`	Conversion factor from bytes to megabytes
`amount_transferred`	`int`	`0`	Number of bytes transferred so far

init_tqdm()

def init_tqdm(self, total_in_mb, position=1):

Parameters

Name	Type	Default	Description
`total_in_mb`	`int \\| float`	Required	Total transfer size in MB
`position`	`int`	`1`	Terminal row position for the progress bar

Behaviour

Creates a tqdm progress bar with:

self.pbar = tqdm(
    desc=self.filename,
    total=total_in_mb,
    unit="MB",
    unit_scale=True,
    position=position,
    leave=False,
)

Example

holder = TQDMHolder("file.csv")
holder.init_tqdm(total_in_mb=100)

callback_s3()

def callback_s3(self, chunk_bytes: int):

Progress callback intended for S3 transfers.

Boto-style callbacks report the size of the most recent chunk, not the cumulative number of bytes transferred. This method accumulates the total internally.

Parameters

Name	Type	Description
`chunk_bytes`	`int`	Number of bytes transferred in the latest callback

Behaviour

Ensures the progress bar exists.
Adds chunk_bytes to amount_transferred.
Updates the progress bar.

callback_scp()

def callback_scp(self, current_bytes: int, total_bytes: int):

Progress callback intended for SCP/SFTP transfers.

SCP-style callbacks usually report the current cumulative byte count and total byte count.

Parameters

Name	Type	Description
`current_bytes`	`int`	Current number of bytes transferred
`total_bytes`	`int`	Total file size in bytes

Behaviour

If the progress bar has not been initialised, it creates one using the total file size.

Then it updates the progress bar to the current byte count.

update()

def update(self, current_bytes: int):

Updates the progress bar to reflect the current transferred byte count.

Parameters

Name	Type	Description
`current_bytes`	`int`	Current number of transferred bytes

Behaviour

Converts bytes to megabytes:

current_mb = int(current_bytes // self.bytes_to_mb)

Then updates the progress bar position:

self.pbar.n = current_mb
self.pbar.refresh()

__enter__()

def __enter__(self):

Context manager entry method.

Returns

The current TQDMHolder instance.

Example

with TQDMHolder("file.csv") as progress:
    progress.init_tqdm(100)

__exit__()

def __exit__(self, exc_type, exc_val, exc_tb):

Context manager exit method.

Closes the progress bar.

Parameters

Name	Description
`exc_type`	Exception type, if raised
`exc_val`	Exception value, if raised
`exc_tb`	Exception traceback, if raised

Behaviour

self.pbar.close()

`irods_adaptor.py`

Overview

irods_adaptor.py implements the iRODS backend for the staging workflow. It defines IRODSAdaptor, a concrete implementation of StorageAdaptor, allowing the workflow to list, read, write, validate, skip, delete, and checksum iRODS data objects.

The adaptor supports the streaming workflow by exposing open_read() and open_write() methods. This allows data to be streamed directly between iRODS and another backend without first staging full files in a temporary local directory.

Imports

import os

from irods.session import iRODSSession

from .base import StorageAdaptor

Import	Purpose
`os`	Reads iRODS connection settings from environment variables and handles path manipulation
`iRODSSession`	Creates an authenticated iRODS session
`StorageAdaptor`	Base adaptor interface implemented by `IRODSAdaptor`

Class: `IRODSAdaptor`

class IRODSAdaptor(StorageAdaptor):

Concrete storage adaptor for iRODS.

This class provides iRODS-specific implementations for:

listing collections and data objects
opening data objects for reading
opening data objects for writing
checking object sizes
checking whether paths exist as data objects or collections
deleting data objects
preparing destinations for streaming writes
retrieving native iRODS checksums

Constructor

__init__()

def __init__(self, prefix):

Parameters

Name	Type	Description
`prefix`	`str`	Environment variable prefix, usually `SRC` or `DST`

Environment Variables

The adaptor expects the following variables:

Variable Pattern	Example	Description
`{prefix}_IRODS_HOST`	`SRC_IRODS_HOST`	iRODS server hostname
`{prefix}_IRODS_PORT`	`SRC_IRODS_PORT`	iRODS server port
`{prefix}_IRODS_USER`	`SRC_IRODS_USER`	iRODS username
`{prefix}_IRODS_PASSWORD`	`SRC_IRODS_PASSWORD`	iRODS password
`{prefix}_IRODS_ZONE`	`SRC_IRODS_ZONE`	iRODS zone

Example

source_adaptor = IRODSAdaptor("SRC")
dest_adaptor = IRODSAdaptor("DST")

For IRODSAdaptor("SRC"), the adaptor reads:

SRC_IRODS_HOST
SRC_IRODS_PORT
SRC_IRODS_USER
SRC_IRODS_PASSWORD
SRC_IRODS_ZONE

Behaviour

initialises:

self.session = iRODSSession(...)

The session is reused by all methods on the adaptor instance.

Path Helper Methods

_sanitize_path()

def _sanitize_path(self, path):

Normalises an iRODS path.

Parameters

Name	Type	Description
`path`	`str`	Raw iRODS path

Returns

A cleaned absolute iRODS path.

Behaviour

The method:

Removes an irods:// prefix if present.
Converts backslashes to forward slashes.
Ensures the path starts with /.
Removes trailing slashes.

Example

self._sanitize_path("irods://tempZone/home/rods/data/")

Returns

/tempZone/home/rods/data

_get_parent()

def _get_parent(self, path):

Returns the parent collection path for an iRODS path.

Parameters

Name	Type	Description
`path`	`str`	iRODS path

Returns

The parent path.

Example

self._get_parent("/tempZone/home/rods/data/file.txt")

Returns

/tempZone/home/rods/data

_ensure_collection_exists()

def _ensure_collection_exists(self, collection_path):

Ensures that an iRODS collection exists, creating parent collections recursively if needed.

Parameters

Name	Type	Description
`collection_path`	`str`	iRODS collection path

Returns

None

Behaviour

The method:

Sanitises the collection path.
Checks whether the collection already exists.
Recursively creates missing parent collections.
Creates the requested collection.

Example

self._ensure_collection_exists("/tempZone/home/rods/output/subdir")

If needed, this creates:

/tempZone/home/rods/output
/tempZone/home/rods/output/subdir

Listing Methods

list_objects()

def list_objects(self, path):

Lists iRODS data objects under a collection or returns a single data object.

This method implements the StorageAdaptor.list_objects() interface.

Parameters

Name	Type	Description
`path`	`str`	iRODS collection path or data object path

Yields

Tuples of:

(relative_path, size_in_bytes)

Item	Type	Description
`relative_path`	`str`	File path relative to the listed root
`size_in_bytes`	`int`	Data object size in bytes

Behaviour

The method first tries to treat path as a collection.

If it is a collection, it recursively yields all data objects under it.

If it is not a collection, it tries to treat path as a single data object.

If neither lookup succeeds, it prints an error.

Collection Example

Given:

/tempZone/home/rods/data/
├── a.csv
└── nested/
    └── b.csv

Calling:

list(adaptor.list_objects("/tempZone/home/rods/data"))

May return:

[
    ("a.csv", 1234),
    ("nested/b.csv", 5678),
]

Single File Example

Calling:

list(adaptor.list_objects("/tempZone/home/rods/data/a.csv"))

May return:

[
    ("a.csv", 1234),
]

Streaming Read/Write Methods

open_read()

def open_read(self, source_path):

Opens an iRODS data object as a readable binary stream.

This method is used when iRODS is the source backend.

**Parameters

Name	Type	Description
`source_path`	`str`	Path to an iRODS data object

Returns

A readable stream from the iRODS data object.

Behaviour

The method:

Sanitises the source path.
Checks whether the path is a collection.
Raises IsADirectoryError if the path is a collection.
Opens the data object in read mode.

Example

with adaptor.open_read("/tempZone/home/rods/data/file.csv") as src:
    chunk = src.read(1024)

Raises

Error	Cause
`IsADirectoryError`	Path points to an iRODS collection
iRODS exception	Data object does not exist or cannot be opened

open_write()

def open_write(self, dest_file_path, overwrite=True):

Opens an iRODS data object as a writable stream.

This method is used when iRODS is the destination backend.

Parameters

Name	Type	Default	Description
`dest_file_path`	`str`	Required	Destination iRODS data object path
`overwrite`	`bool`	`True`	Whether to overwrite an existing data object

Returns

A writable stream for the iRODS data object.

Behaviour

The method:

Sanitises the destination path.
Determines the parent collection.
Creates missing parent collections.
Checks whether the destination path already exists as a collection.
Deletes an existing data object if overwrite=True.
Raises FileExistsError if the object exists and overwrite=False.
Creates a new iRODS data object.
Opens it in write mode.

Example

with adaptor.open_write("/tempZone/home/rods/output/file.csv") as dst:
    dst.write(b"example data")

Raises

Error	Cause
`IsADirectoryError`	Destination path exists as a collection
`FileExistsError`	Destination object exists and `overwrite=False`
iRODS exception	Object cannot be created or opened

Size and Existence Methods

get_size()

def get_size(self, path):

Returns the size of an iRODS data object in bytes.

This method implements the StorageAdaptor.get_size() interface.

Parameters

Name	Type	Description
`path`	`str`	iRODS data object path

Returns

The data object size in bytes.

Example

size = adaptor.get_size("/tempZone/home/rods/data/file.csv")

Returns

Raises

An iRODS exception if the object does not exist or cannot be accessed.

exists_as_data_object()

def exists_as_data_object(self, path):

Checks whether a path exists as an iRODS data object.

Parameters

Name	Type	Description
`path`	`str`	iRODS path

Returns

Return Value	Meaning
`True`	Path exists as a data object
`False`	Path does not exist as a data object

Example

adaptor.exists_as_data_object("/tempZone/home/rods/data/file.csv")

exists_as_collection()

def exists_as_collection(self, path):

Checks whether a path exists as an iRODS collection.

Parameters

Name	Type	Description
`path`	`str`	iRODS path

Returns

Return Value	Meaning
`True`	Path exists as a collection
`False`	Path does not exist as a collection

Example

adaptor.exists_as_collection("/tempZone/home/rods/data")

Delete and Resume Methods

delete_data_object()

def delete_data_object(self, path):

Deletes an iRODS data object if it exists.

Parameters

Name	Type	Description
`path`	`str`	iRODS data object path

Returns

None

Behaviour

The method:

Sanitises the path.
Checks whether it exists as a data object.
Deletes it using force=True if it exists.

Example

adaptor.delete_data_object("/tempZone/home/rods/output/file.csv")

prepare_for_stream_write()

def prepare_for_stream_write(self, dest_file_path, expected_size=None):

Prepares an iRODS destination path before streaming a file.

This method supports file-level resume/skip behaviour.

Parameters

Name	Type	Default	Description
`dest_file_path`	`str`	Required	Destination iRODS data object path
`expected_size`	`int \\| None`	`None`	Expected file size in bytes

Returns

Return Value	Meaning
`True`	Existing destination object has the expected size and can be skipped
`False`	Transfer should proceed

Behaviour

The method:

Sanitises the destination path.
Raises an error if the destination exists as a collection.
Returns False if no data object exists at the destination.
If a data object exists and its size matches expected_size, returns True.
If a data object exists but its size differs or size is unknown, deletes it and returns False.

Example

should_skip = adaptor.prepare_for_stream_write(
    "/tempZone/home/rods/output/file.csv",
    expected_size=1048576,
)

if should_skip:
    print("File already exists and is complete")

Raises

Error	Cause
`IsADirectoryError`	Destination exists as an iRODS collection

Checksum Methods

get_remote_checksum()

def get_remote_checksum(self, path):

Returns the native iRODS checksum for a data object.

Parameters

Name	Type	Description
`path`	`str`	iRODS data object path

Returns

A checksum string, or None if the checksum cannot be retrieved.

Behaviour

The method:

Sanitises the path.
Retrieves the iRODS data object.
Uses the existing registered checksum if present.
Otherwise asks iRODS to calculate one with chksum().
Removes a checksum algorithm prefix if present.

For example, if iRODS returns:

sha2:abc123

the method returns:

abc123

Example

checksum = adaptor.get_remote_checksum("/tempZone/home/rods/output/file.csv")

Error Handling

If anything fails, the method returns:

None

instead of raising the exception.

Destructor**

__del__()

def __del__(self):

Cleans up the iRODS session when the adaptor instance is garbage collected.

Behaviour

If the adaptor has a session attribute, the method calls:

self.session.cleanup()

Environment Configuration Example

[env]
SOURCE_TYPE=IRODS
DEST_TYPE=FS
SOURCE_PATH=/tempZone/home/rods/input
DEST_PATH=/tmp/output

[creds]
SRC_IRODS_HOST=vm-10-195-10-192.cloud.mwn.de
SRC_IRODS_PORT=1247
SRC_IRODS_USER=rods
SRC_IRODS_PASSWORD=your-password
SRC_IRODS_ZONE=tempZone

For iRODS as a destination:

[env]
SOURCE_TYPE=S3
DEST_TYPE=IRODS
SOURCE_PATH=/Cleaned_Data_EDD
DEST_PATH=/tempZone/home/rods/output

[creds]
DST_IRODS_HOST=vm-10-195-10-192.cloud.mwn.de
DST_IRODS_PORT=1247
DST_IRODS_USER=rods
DST_IRODS_PASSWORD=your-password
DST_IRODS_ZONE=tempZone

Example Usage

iRODS as Source

adaptor = IRODSAdaptor("SRC")

for rel_path, size in adaptor.list_objects("/tempZone/home/rods/input"):
    print(rel_path, size)

with adaptor.open_read("/tempZone/home/rods/input/file.csv") as src:
    data = src.read(1024)

iRODS as Destination

adaptor = IRODSAdaptor("DST")

with adaptor.open_write("/tempZone/home/rods/output/file.csv") as dst:
    dst.write(b"example data")

size = adaptor.get_size("/tempZone/home/rods/output/file.csv")

`s3_adaptor.py`

Overview

s3_adaptor.py implements the S3-compatible backend for the staging workflow. It defines S3Adaptor, a concrete implementation of StorageAdaptor, allowing the workflow to list, read, upload, size-check, skip, delete, and checksum objects in S3-compatible object storage.

The adaptor is designed for streaming transfers. It can stream data from S3 using open_read() and stream data to S3 using upload_stream(). Since S3 does not expose normal writable file handles, open_write() is intentionally unsupported.

This adaptor works with AWS S3 and S3-compatible services such as MinIO, provided the correct endpoint and credentials are configured.

Imports

import logging
import os
from pathlib import Path

import boto3
import botocore
from boto3.s3.transfer import TransferConfig
from botocore.client import Config
from types_boto3_s3.client import S3Client

from .base import StorageAdaptor, TQDMHolder

Import	Purpose
`logging`	Logs S3 operations
`os`	Reads S3 configuration from environment variables
`boto3`	Creates the S3 client
`botocore`	Handles S3 client errors
`TransferConfig`	Configures multipart upload behaviour
`Config`	Configures S3 client options
`S3Client`	Type hint for the boto3 S3 client
`StorageAdaptor`	Base adaptor interface
`Path`	Imported but not currently used
`TQDMHolder`	Imported but not currently used in active code

Logger

logging.basicConfig(format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

Initialises module-level logging for S3 adaptor operations.

The adaptor logs actions such as:

opening S3 objects for reading
streaming uploads to S3
skipping existing objects
deleting incomplete or mismatched objects

Class: `S3Adaptor`

class S3Adaptor(StorageAdaptor):

Concrete storage adaptor for S3-compatible object storage.

This class implements the streaming interface required by the staging workflow.

Supported operations include:

listing objects under a prefix
opening an object as a readable stream
uploading a readable stream to S3
retrieving object size
checking existing destination objects
retrieving MD5 metadata or safe ETag checksums

Constructor

__init__()

def __init__(
    self,
    prefix,
    transfer_config: TransferConfig = None,
    position: int = 1,
):

Initialises an S3 adaptor using environment variables and creates a boto3 S3 client.

Parameters

Name	Type	Default	Description
`prefix`	`str`	Required	Environment variable prefix, usually `SRC` or `DST`
`transfer_config`	`TransferConfig \\| None`	`None`	Optional boto3 transfer configuration
`position`	`int`	`1`	Progress-bar position used by worker processes

Environment Variables

The adaptor expects the following variables:

Variable Pattern	Example	Required	Description
`{prefix}_S3_BUCKET`	`SRC_S3_BUCKET`	Yes	S3 bucket name
`{prefix}_S3_ENDPOINT_URL`	`SRC_S3_ENDPOINT_URL`	Yes	S3 or MinIO endpoint URL
`{prefix}_S3_ACCESS_KEY_ID`	`SRC_S3_ACCESS_KEY_ID`	Yes	S3 access key
`{prefix}_S3_SECRET_ACCESS_KEY`	`SRC_S3_SECRET_ACCESS_KEY`	Yes	S3 secret key
`{prefix}_S3_REGION`	`SRC_S3_REGION`	No	S3 region, defaults to `us-east-1`
`{prefix}_S3_VERIFY_SSL`	`SRC_S3_VERIFY_SSL`	No	Whether to verify SSL certificates, defaults to `true`

Example

source_adaptor = S3Adaptor("SRC")
dest_adaptor = S3Adaptor("DST")

For S3Adaptor("SRC"), the adaptor reads:

SRC_S3_BUCKET
SRC_S3_ENDPOINT_URL
SRC_S3_ACCESS_KEY_ID
SRC_S3_SECRET_ACCESS_KEY
SRC_S3_REGION
SRC_S3_VERIFY_SSL

S3 Client Configuration

The adaptor creates a boto3 client with:

s3_config = Config(
    signature_version="s3v4",
    s3={"addressing_style": "path"},
    connect_timeout=60,
    read_timeout=300,
    retries={"mode": "adaptive", "total_max_attempts": 10},
)

Behaviour

Setting	Purpose
`signature_version="s3v4"`	Uses AWS Signature Version 4
`addressing_style="path"`	Uses path-style bucket addressing, useful for MinIO
`connect_timeout=60`	Allows up to 60 seconds for connection establishment
`read_timeout=300`	Allows up to 300 seconds for read operations
`retries={"mode": "adaptive", "total_max_attempts": 10}`	Enables adaptive retries for transient failures

SSL Verification

verify_env = os.getenv(f"{prefix}_S3_VERIFY_SSL", "true").strip().lower()
verify_ssl = verify_env not in {"0", "false", "no"}

SSL verification is enabled by default.

To disable SSL verification for testing or internal MinIO endpoints:

SRC_S3_VERIFY_SSL=false

or:

DST_S3_VERIFY_SSL=false

Transfer Configuration

If no custom TransferConfig is supplied, the adaptor uses:

Setting	Value	Description
`multipart_threshold`	`64 MiB`	Objects larger than this use multipart upload
`multipart_chunksize`	`1 GiB`	Size of each multipart upload part
`max_concurrency`	`2`	Number of S3 upload threads per transfer
`use_threads`	`True`	Enables threaded multipart uploads

Path Helper Methods

_sanitize_path()

def _sanitize_path(self, path):

Normalises an S3 path into an object key.

Parameters

Name	Type	Description
`path`	`str`	Raw S3 path, object key, or `s3://` URI

Returns

A clean S3 object key without a leading slash or bucket prefix.

Behaviour

The method:

Removes an s3:// prefix if present.
Removes the bucket name from the start of the path if present.
Converts backslashes to forward slashes.
Removes leading slashes.

Examples

self._sanitize_path("s3://my-bucket/data/file.csv")

Returns:

data/file.csv

self._sanitize_path("/data/file.csv")

Returns:

data/file.csv

self._sanitize_path("data\\file.csv")

Returns

data/file.csv

Listing Methods

list_objects()

def list_objects(self, path):

Lists S3 objects under a prefix or returns a single object if the path points directly to an object.

This method implements the StorageAdaptor.list_objects() interface.

Parameters

Name	Type	Description
`path`	`str`	S3 prefix, object key, or `s3://` URI

Yields

Tuples of:

(relative_path, size_in_bytes)

Item	Type	Description
`relative_path`	`str`	Object path relative to the requested prefix
`size_in_bytes`	`int`	Object size in bytes

Behaviour

The method first checks whether path is a single object:

response = self.client.head_object(Bucket=self.bucket, Key=prefix)

If that succeeds, it yields one manifest item:

yield os.path.basename(prefix), response["ContentLength"]

If the object does not exist, it treats the path as a prefix and paginates through matching objects:

paginator = self.client.get_paginator("list_objects_v2")

Each listed object is returned relative to the prefix.

Single Object Example

Given:

s3://my-bucket/data/file.csv

Calling:

list(adaptor.list_objects("data/file.csv"))

May return:

[
    ("file.csv", 1048576),
]

Prefix Example

Given:

s3://my-bucket/data/
├── a.csv
└── nested/b.csv

Calling:

list(adaptor.list_objects("data"))

May return:

[
    ("a.csv", 1234),
    ("nested/b.csv", 5678),
]

Empty Prefix Example

Calling:

list(adaptor.list_objects(""))

lists objects from the bucket root.

Error Handling

If head_object() fails with one of these codes, the method treats the path as a prefix:

404
NoSuchKey
NotFound

Other errors are re-raised.

Streaming Read/Write Methods

open_read()

def open_read(self, source_path):

Opens an S3 object as a readable binary stream.

This method is used when S3 is the source backend. Parameters

Name	Type	Description
`source_path`	`str`	S3 object key or path

Returns

A readable streaming body returned by boto3:

response["Body"]

This object supports:

read(size)
close()

Example

with adaptor.open_read("data/file.csv") as src:
    chunk = src.read(8192)

Behaviour

The method:

Sanitises the source path.
Logs the object key.
Calls get_object().
Returns the response body stream.

response = self.client.get_object(
    Bucket=self.bucket,
    Key=key,
)

Raises

Boto3 or botocore exceptions if the object cannot be opened.

open_write()

def open_write(self, dest_file_path, overwrite=True):

S3 does not support normal writable file handles.

This method always raises NotImplementedError.

Parameters

Name	Type	Default	Description
`dest_file_path`	`str`	Required	Destination object path
`overwrite`	`bool`	`True`	Included for interface compatibility

Raises

NotImplementedError

Message

S3 does not support open_write(). Use upload_stream() instead.

upload_stream()

def upload_stream(self, stream, dest_file_path, extra_args=None):

Uploads a readable binary stream directly to S3.

This method is used when S3 is the destination backend.

Parameters

Name	Type	Default	Description
`stream`	file-like object	Required	Readable binary stream
`dest_file_path`	`str`	Required	Destination S3 object key or path
`extra_args`	`dict \\| None`	`None`	Optional boto3 upload arguments

Returns

None

Behaviour

The method:

Sanitises the destination path.
Logs the target key.
Calls upload_fileobj().

self.client.upload_fileobj(
    stream,
    self.bucket,
    key,
    ExtraArgs=extra_args or {},
    Config=self.transfer_config,
)

Example

with source_adaptor.open_read("/source/file.csv") as src:
    s3_adaptor.upload_stream(src, "output/file.csv")

Raises

Boto3 or botocore exceptions if the upload fails.

Size and Resume Methods

get_size()

Returns the size of an S3 object in bytes.

This method implements the StorageAdaptor.get_size() interface.

Parameters

Name	Type	Description
`path`	`str`	S3 object key or path

Returns

Object size in bytes.

Behaviour

The method:

Sanitises the path.
Calls head_object().
Returns ContentLength.

response = self.client.head_object(
    Bucket=self.bucket,
    Key=key,
)

return response["ContentLength"]

Example

size = adaptor.get_size("data/file.csv")

Returns:

Raises

Boto3 or botocore exceptions if the object does not exist or cannot be accessed.

prepare_for_stream_write()

def prepare_for_stream_write(self, dest_file_path, expected_size=None):

Prepares an S3 destination before streaming upload.

This method supports file-level resume/skip behaviour.

Parameters

Name	Type	Default	Description
`dest_file_path`	`str`	Required	Destination S3 object key or path
`expected_size`	`int \\| None`	`None`	Expected object size in bytes

Returns

Return Value	Meaning
`True`	Existing object has the expected size and can be skipped
`False`	Upload should proceed

Behaviour

The method:

Sanitises the destination key.
Checks if the destination object exists with head_object().
If the object exists and its size matches expected_size, returns True.
If the object exists but size differs, deletes it and returns False.
If the object does not exist, returns False.
Re-raises unexpected S3 errors.

Matching Existing Object

if expected_size is not None and existing_size == expected_size:
    logger.info("[S3] Existing object has matching size, skipping: %s", key)
    return True

Mismatched Existing Object

self.client.delete_object(
    Bucket=self.bucket,
    Key=key,
)

Not Found Codes

The following errors are treated as “object does not exist”:

404
NoSuchKey
NotFound

Example

should_skip = adaptor.prepare_for_stream_write(
    "output/file.csv",
    expected_size=1048576,
)

if not should_skip:
    adaptor.upload_stream(src, "output/file.csv")

Checksum Methods

def get_remote_checksum(self, remote_path):

Returns an S3 object checksum if one can be safely determined.

Parameters

Name	Type	Description
`remote_path`	`str`	S3 object key or path

Returns

A checksum string, or None.

Behaviour

The method:

Sanitises the object key.
Calls head_object().
Checks object metadata for an md5 value.
If metadata contains md5, returns that value.
Otherwise, reads the object ETag.
Returns the ETag only if it does not contain -.
Returns None for multipart ETags or on failure.

metadata = response.get("Metadata", {})

if "md5" in metadata:
    return metadata["md5"]

etag = response.get("ETag", "").replace('"', "")

return etag if "-" not in etag else None

Example Environment Configuration

S3 as Source

[env]
SOURCE_TYPE=S3
DEST_TYPE=IRODS
SOURCE_PATH=/Cleaned_Data_EDD
DEST_PATH=/tempZone/home/rods/Cleaned_Data_EDD

[creds]
SRC_S3_ENDPOINT_URL=https://minio.example.org:9000
SRC_S3_ACCESS_KEY_ID=your-access-key
SRC_S3_SECRET_ACCESS_KEY=your-secret-key
SRC_S3_REGION=us-east-1
SRC_S3_BUCKET=wp6-health
SRC_S3_VERIFY_SSL=false

S3 as Destination

[env]
SOURCE_TYPE=IRODS
DEST_TYPE=S3
SOURCE_PATH=/tempZone/home/rods/Cleaned_Data_EDD
DEST_PATH=/backup/Cleaned_Data_EDD

[creds]
DST_S3_ENDPOINT_URL=https://minio.example.org:9000
DST_S3_ACCESS_KEY_ID=your-access-key
DST_S3_SECRET_ACCESS_KEY=your-secret-key
DST_S3_REGION=us-east-1
DST_S3_BUCKET=backup-bucket
DST_S3_VERIFY_SSL=false

Example Usage

S3 as Source

adaptor = S3Adaptor("SRC")

for rel_path, size in adaptor.list_objects("Cleaned_Data_EDD"):
    print(rel_path, size)

with adaptor.open_read("Cleaned_Data_EDD/population_malta.json") as src:
    chunk = src.read(1024)

S3 as Destination

adaptor = S3Adaptor("DST")

with open("population_malta.json", "rb") as src:
    adaptor.upload_stream(src, "backup/population_malta.json")

size = adaptor.get_size("backup/population_malta.json")

`fs_adaptor.py`

Overview

fs_adaptor.py implements the local filesystem backend for the staging workflow. It defines FSAdaptor, a concrete implementation of StorageAdaptor, allowing the workflow to list, read, write, validate, skip, and checksum files on a local or mounted filesystem.

The adaptor is used when either the source or destination type is configured as:

SOURCE_TYPE=FS

or:

DEST_TYPE=FS

It supports the streaming workflow by exposing normal file-based open_read() and open_write() methods. This makes it suitable for transfers such as:

FS -> S3
FS -> iRODS
S3 -> FS
iRODS -> FS
SCP -> FS

Class: `FSAdaptor`

class FSAdaptor(StorageAdaptor):

Concrete storage adaptor for local filesystem paths.

The adaptor provides filesystem-specific implementations for:

listing files under a directory
opening files for binary reading
opening files for binary writing
checking file sizes
creating destination directories
skipping already-complete destination files
calculating local checksums

Constructor

__init__()

def __init__(self, prefix):

Initialises the filesystem adaptor.

Parameter

Name	Type	Description
`prefix`	`str`	Environment variable prefix, usually `SRC` or `DST`

Behaviour

The filesystem adaptor usually does not require credentials, but it still accepts a prefix for consistency with the other adaptors.

Example:

source_adaptor = FSAdaptor("SRC")
dest_adaptor = FSAdaptor("DST")

Unlike S3, iRODS, or SCP, the filesystem adaptor typically does not need variables such as usernames, passwords, hosts, or buckets.

Path Handling

Filesystem paths are handled as local paths.

Examples:

/data/input/file.csv
/tmp/staging/output
/dss/dsshome1/05/di38jim2/data

The adaptor should preserve normal filesystem paths and use Python’s pathlib.Path or os.path utilities to resolve directories, parents, and file sizes.

Listing Methods

list_objects()

def list_objects(self, path: str):

Lists files under a local directory or returns a single file.

This method implements the StorageAdaptor.list_objects() interface.

Parameters

Name	Type	Description
`path`	`str`	Local file or directory path

Yields

Tuples of:

(relative_path, size_in_bytes)

Item	Type	Description
`relative_path`	`str`	File path relative to the source root
`size_in_bytes`	`int`	File size in bytes

Behaviour

If path is a file, the method yields a single item:

("file.csv", 1048576)

If path is a directory, the method recursively walks the directory and yields all files under it.

For example, given:

/data/input/
├── a.csv
└── nested/
    └── b.csv

Calling:

list(adaptor.list_objects("/data/input"))

should return:

[
    ("a.csv", 1234),
    ("nested/b.csv", 5678),
]

Expected Implementation Pattern

def list_objects(self, path: str):
    root = Path(path)

    if root.is_file():
        yield root.name, root.stat().st_size
        return

    for file_path in root.rglob("*"):
        if file_path.is_file():
            rel_path = file_path.relative_to(root)
            yield str(rel_path), file_path.stat().st_size

Raises

Error	Cause
`FileNotFoundError`	Source path does not exist
`NotADirectoryError`	Path is invalid for directory listing
`PermissionError`	User cannot access the file or directory

Streaming Read/Write Methods

open_read()

def open_read(self, source_path: str):

Opens a local file as a readable binary stream.

This method is used when the filesystem is the source backend.

Parameters

Name	Type	Description
`source_path`	`str`	Local source file path

Returns

A readable binary file object.

Example

with adaptor.open_read("/data/input/file.csv") as src:
    chunk = src.read(1024)

Expected Implementation

def open_read(self, source_path: str):
    return open(source_path, "rb")

Raises

Error	Cause
`FileNotFoundError`	Source file does not exist
`IsADirectoryError`	Source path is a directory
`PermissionError`	File cannot be read

open_write()

def open_write(self, dest_path: str, overwrite: bool = True):

Opens a local destination file as a writable binary stream.

This method is used when the filesystem is the destination backend.

Parameters

Name	Type	Default	Description
`dest_path`	`str`	Required	Local destination file path
`overwrite`	`bool`	`True`	Whether to overwrite an existing file

Returns

A writable binary file object.

Behaviour

The method should:

Resolve the destination path.
Create missing parent directories.
Open the file in binary write mode.
Respect the overwrite setting.

Example

with adaptor.open_write("/tmp/output/file.csv", overwrite=True) as dst:
    dst.write(b"example data")

Raises

Error	Cause
`FileExistsError`	Destination exists and `overwrite=False`
`IsADirectoryError`	Destination path points to a directory
`PermissionError`	Destination cannot be written

Size Methods

get_size()

def get_size(self, path: str) -> int:

Returns the size of a local file in bytes.

This method implements the StorageAdaptor.get_size() interface.

Parameters

Name	Type	Description
`path`	`str`	Local file path

Returns

File size in bytes.

Example

size = adaptor.get_size("/tmp/output/file.csv")

Returns:

Raises

Error	Cause
`FileNotFoundError`	File does not exist
`IsADirectoryError`	Path points to a directory
`PermissionError`	File metadata cannot be accessed

Resume and Skip Methods

prepare_for_stream_write()

def prepare_for_stream_write(
    self,
    dest_path: str,
    expected_size: int | None = None,
) -> bool:

Prepares a local filesystem destination before streaming a file.

This method supports file-level resume/skip behaviour.

Parameters

Return Value	Meaning
`True`	Existing file has the expected size and can be skipped
`False`	Transfer should proceed

Returns

Return Value	Meaning
`True`	Existing file has the expected size and can be skipped
`False`	Transfer should proceed

Behaviour

The method should:

Check whether the destination exists.
If it does not exist, return False.
If it exists as a directory, raise IsADirectoryError.
If it exists and size matches expected_size, return True.
If it exists and size differs, delete the file and return False.

Example

should_skip = adaptor.prepare_for_stream_write(
    "/tmp/output/file.csv",
    expected_size=1048576,
)

if should_skip:
    print("File already exists and is complete")

Checksum Methods

get_remote_checksum()

def get_remote_checksum(self, path: str):

Returns the checksum of a local file.

Although the name says “remote”, for the filesystem adaptor this refers to the local file at the given path.

Parameters

Name	Type	Description
`path`	`str`	Local file path

Returns

Checksum as a string, usually MD5.

Example

checksum = adaptor.get_remote_checksum("/tmp/output/file.csv")

Returns

8747a4309a6e27e05334b71ff2c9894e

Example Environment Configuration

Filesystem as Source

[env]
SOURCE_TYPE=FS
DEST_TYPE=S3
SOURCE_PATH=/dss/dsshome1/05/di38jim2/input
DEST_PATH=/backup/input

CONCURRENCY=4
STREAM_CHUNK_SIZE_MB=64
CHECKSUM_STRATEGY=FULL
CHECKSUM_ALGORITHM=md5

No [creds] entries are required for the filesystem source.

Filesystem as Destination

[env]
SOURCE_TYPE=S3
DEST_TYPE=FS
SOURCE_PATH=/Cleaned_Data_EDD
DEST_PATH=/dss/dsshome1/05/di38jim2/output

CONCURRENCY=4
STREAM_CHUNK_SIZE_MB=64
CHECKSUM_STRATEGY=FULL
CHECKSUM_ALGORITHM=md5

Example Usage

List Local Files

adaptor = FSAdaptor("SRC")

for rel_path, size in adaptor.list_objects("/data/input"):
    print(rel_path, size)

Example output:

a.csv 1234
nested/b.csv 5678

Read a Local File

adaptor = FSAdaptor("SRC")

with adaptor.open_read("/data/input/a.csv") as src:
    chunk = src.read(1024)

Write a Local File

adaptor = FSAdaptor("DST")

with adaptor.open_write("/tmp/output/a.csv", overwrite=True) as dst:
    dst.write(b"example data")

Check Destination Size

adaptor = FSAdaptor("DST")

size = adaptor.get_size("/tmp/output/a.csv")
print(size)

Skip Existing Complete File

adaptor = FSAdaptor("DST")

should_skip = adaptor.prepare_for_stream_write(
    "/tmp/output/a.csv",
    expected_size=1234,
)

if should_skip:
    print("Skipping existing complete file")

`scp_adaptor.py`

Overview

scp_adaptor.py implements the SCP/SFTP backend for the staging workflow. It defines SCPAdaptor, a concrete implementation of StorageAdaptor, using paramiko to connect to a remote host over SSH and transfer files through SFTP streams.

The adaptor supports the streaming workflow by exposing:

open_read()
open_write()
list_objects()
get_size()
prepare_for_stream_write()
get_remote_checksum()

This allows the workflow to stream files:

SCP -> FS
SCP -> S3
SCP -> iRODS
FS -> SCP
S3 -> SCP
iRODS -> SCP

Imports

import logging
import os
import stat

import paramiko

from .base import StorageAdaptor, TQDMHolder

Import	Purpose
`logging`	Logs SCP/SFTP operations and errors
`os`	Reads environment variables and handles path operations
`stat`	Checks whether remote paths are files or directories
`paramiko`	Provides SSH and SFTP functionality
`StorageAdaptor`	Base adaptor interface
`TQDMHolder`	Imported but not currently used in active code

Logger

logging.basicConfig(format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

Initialises module-level logging.

Used for messages such as:

missing remote paths
skipped existing files
deleting incomplete or mismatched files

Class: `SCPAdaptor`

class SCPAdaptor(StorageAdaptor):

Concrete storage adaptor for remote SSH/SFTP-accessible filesystems.

The adaptor provides methods for:

connecting to a remote host
listing files recursively
opening remote files for reading
opening remote files for writing
checking remote file sizes
skipping already-complete destination files
deleting incomplete destination files
calculating remote MD5 checksums using md5sum
closing SSH/SFTP connections

Constructor

__init__()

Creates an SSH connection and opens an SFTP session.

Parameters

Name	Type	Description
`prefix`	`str`	Environment variable prefix, usually `SRC` or `DST`

Environment Variables

The adaptor expects the following variables:

Variable Pattern	Example	Required	Description
`{prefix}_SCP_HOST`	`SRC_SCP_HOST`	Yes	Remote SSH hostname or IP address
`{prefix}_SCP_PORT`	`SRC_SCP_PORT`	No	SSH port, defaults to `22`
`{prefix}_SCP_USER`	`SRC_SCP_USER`	Yes	SSH username
`{prefix}_SCP_PASSWORD`	`SRC_SCP_PASSWORD`	Yes	SSH password

Example

source_adaptor = SCPAdaptor("SRC")
dest_adaptor = SCPAdaptor("DST")

For SCPAdaptor("SRC"), the adaptor reads:

SRC_SCP_HOST
SRC_SCP_PORT
SRC_SCP_USER
SRC_SCP_PASSWORD

SSH Setup

self.ssh = paramiko.SSHClient()
self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

The adaptor creates a Paramiko SSH client and automatically accepts missing host keys.

Connection

self.ssh.connect(
    self.host,
    port=self.port,
    username=self.user,
    password=self.password,
)
self.sftp = self.ssh.open_sftp()

If the connection fails, the constructor raises:

ConnectionError

with the message:

SCP Connection Failed: ...

Listing Methods

list_objects()

def list_objects(self, path):

Lists files under a remote path.

This method implements the StorageAdaptor.list_objects() interface.

Parameters

Name	Type	Description
`path`	`str`	Remote file or directory path

Yields

Tuples of:

(relative_path, size_in_bytes)

Item	Type	Description
`relative_path`	`str`	File path relative to the listed root
`size_in_bytes`	`int`	Remote file size in bytes

Behaviour

The method first normalises the path:

path = path.replace("\\", "/")

Then it checks the remote path type:

attr = self.sftp.stat(path)

If the path is a regular file:

yield os.path.basename(path), attr.st_size

If the path is a directory, it recursively walks through the directory using:

self.sftp.listdir_attr(base_path)

and yields all files below it.

Single File Example

Given this remote file:

/home/user/data/file.csv

Calling:

list(adaptor.list_objects("/home/user/data/file.csv"))

May return:

[
    ("file.csv", 1048576),
]

Directory Example

Given this remote directory:

/home/user/data/
├── a.csv
└── nested/
    └── b.csv

Calling:

list(adaptor.list_objects("/home/user/data"))

May return:

[
    ("a.csv", 1234),
    ("nested/b.csv", 5678),
]

Error Handling

If the path does not exist, the method logs:

[SCP] Path not found: <path>

and returns without yielding anything.

Streaming Read/Write Methods

open_read()

def open_read(self, source_path):

Opens a remote SCP/SFTP file as a readable binary stream.

This method is used when SCP/SFTP is the source backend.

Parameters

Name	Type	Description
`source_path`	`str`	Remote source file path

Returns

A Paramiko SFTP file object opened in binary read mode.

self.sftp.open(remote_path, "rb")

Behavior

The method:

Normalises path separators.
Checks the remote path with sftp.stat().
Verifies that the path is a regular file.
Opens the remote file for binary reading.

Example

with adaptor.open_read("/home/user/data/file.csv") as src:
    chunk = src.read(1024)

Raises

Error	Cause
`IsADirectoryError`	Source path is not a regular file
`FileNotFoundError`	Source file does not exist
Paramiko/SFTP exception	Remote file cannot be accessed

open_write()

def open_write(self, dest_file_path, overwrite=True):

Opens a remote SCP/SFTP file as a writable binary stream.

This method is used when SCP/SFTP is the destination backend.

Parameters

Name	Type	Default	Description
`dest_file_path`	`str`	Required	Remote destination file path
`overwrite`	`bool`	`True`	Whether to overwrite an existing file

Returns

A Paramiko SFTP file object opened in binary write mode.

self.sftp.open(remote_path, "wb")

Behaviour

The method:

Normalises path separators.
Determines the parent directory.
Creates the parent directory if needed.
Checks whether the destination already exists.
Raises an error if the destination is a directory.
Raises an error if the destination exists and overwrite=False.
Removes the existing file if overwrite=True.
Opens the destination file for binary writing.

Example

with adaptor.open_write("/home/user/output/file.csv", overwrite=True) as dst:
    dst.write(b"example data")

Raises

Error	Cause
`IsADirectoryError`	Destination exists as a directory
`FileExistsError`	Destination exists and `overwrite=False`
`FileNotFoundError`	Parent path cannot be created or accessed
Paramiko/SFTP exception	Remote file cannot be opened

Size Methods

get_size()

def get_size(self, path):

Returns the size of a remote SCP/SFTP file in bytes.

This method implements the StorageAdaptor.get_size() interface.

Parameters

Name	Type	Description
`path`	`str`	Remote file path

Returns

Remote file size in bytes.

return self.sftp.stat(remote_path).st_size

Example

size = adaptor.get_size("/home/user/output/file.csv")

Returns

Raises

Error	Cause
`FileNotFoundError`	Remote file does not exist
Paramiko/SFTP exception	Remote metadata cannot be read

Resume and Skip Methods

prepare_for_stream_write()

Prepares a remote SCP/SFTP destination path before streaming upload.

This method supports file-level resume/skip behaviour.

Parameters

Name	Type	Default	Description
`dest_file_path`	`str`	Required	Remote destination file path
`expected_size`	`int \\| None`	`None`	Expected file size in bytes

Returns

Return Value	Meaning
`True`	Existing destination file has the expected size and can be skipped
`False`	Transfer should proceed

Behaviour

The method:

Normalises the destination path.
Checks whether the destination exists.
Raises an error if the destination is a directory.
If the existing file size matches expected_size, returns True.
If the existing file size differs, deletes the file and returns False.
If the file does not exist, returns False.

Matching Existing File

if expected_size is not None and existing_size == expected_size:
    logger.info("[SCP] Existing file has matching size, skipping: %s", remote_path)
    return True

Mismatched Existing File

self.sftp.remove(remote_path)
return False

Example

should_skip = adaptor.prepare_for_stream_write(
    "/home/user/output/file.csv",
    expected_size=1048576,
)

if should_skip:
    print("Skipping existing complete file")

Destructor

__del__()

def __del__(self):

Closes the SFTP and SSH connections when the adaptor object is garbage collected.

Behaviour

If self.sftp exists, the method attempts to close it:

self.sftp.close()

If self.ssh exists, the method attempts to close it:

self.ssh.close()

OSError exceptions are ignored during cleanup.

Environment Configuration Example

SCP/SFTP as Source

[env]
SOURCE_TYPE=SCP
DEST_TYPE=S3
SOURCE_PATH=/home/user/input
DEST_PATH=/backup/input

CONCURRENCY=4
STREAM_CHUNK_SIZE_MB=64
CHECKSUM_STRATEGY=FULL
CHECKSUM_ALGORITHM=md5

[creds]
SRC_SCP_HOST=remote.example.org
SRC_SCP_PORT=22
SRC_SCP_USER=myuser
SRC_SCP_PASSWORD=mypassword

SCP/SFTP as Destination

[env]
SOURCE_TYPE=S3
DEST_TYPE=SCP
SOURCE_PATH=/Data
DEST_PATH=/home/user/output/Data

CONCURRENCY=4
STREAM_CHUNK_SIZE_MB=64
CHECKSUM_STRATEGY=FULL
CHECKSUM_ALGORITHM=md5

[creds]
DST_SCP_HOST=remote.example.org
DST_SCP_PORT=22
DST_SCP_USER=myuser
DST_SCP_PASSWORD=mypassword

Example Usage

SCP/SFTP as Source

adaptor = SCPAdaptor("SRC")

for rel_path, size in adaptor.list_objects("/home/user/input"):
    print(rel_path, size)

with adaptor.open_read("/home/user/input/file.csv") as src:
    chunk = src.read(1024)

SCP/SFTP as Destination

adaptor = SCPAdaptor("DST")

with adaptor.open_write("/home/user/output/file.csv", overwrite=True) as dst:
    dst.write(b"example data")

size = adaptor.get_size("/home/user/output/file.csv")

Skip Existing Complete File

should_skip = adaptor.prepare_for_stream_write(
    "/home/user/output/file.csv",
    expected_size=2048,
)

if should_skip:
    print("Remote file already exists with matching size")

API Reference

cli.py

Functions

Required Environment Variables

Optional Environment Variables

Checksum Strategies

Note

Classes

ProgressReader

Constructor

StreamingFileRunner

[env]

[creds]

Command-Line Usage

Validation Behaviour

Size Validation

Full Checksum Validation

Error Handling

Backend Requirements

Required for Source Adaptors

Required for Destination Adaptors

Example Transfer Flow

Adaptors

base.py

Overview

Class: StorageAdaptor

Constructor

Abstract Methods

Class: TQDMHolder

irods_adaptor.py

Class: IRODSAdaptor

Constructor

Path Helper Methods

Listing Methods

Streaming Read/Write Methods

Size and Existence Methods

Delete and Resume Methods

Checksum Methods

Destructor**

Environment Configuration Example

s3_adaptor.py

Class: S3Adaptor

Constructor

S3 Client Configuration

Path Helper Methods

Streaming Read/Write Methods

Checksum Methods

Example Environment Configuration

S3 as Source

S3 as Destination

fs_adaptor.py

Class: FSAdaptor

Constructor

Size Methods

Resume and Skip Methods

Checksum Methods

Example Environment Configuration

Filesystem as Source

Filesystem as Destination

scp_adaptor.py

Class: SCPAdaptor

Constructor

Streaming Read/Write Methods

Size Methods

Resume and Skip Methods

Destructor

Environment Configuration Example

`cli.py`

`ProgressReader`

`StreamingFileRunner`

`[env]`

`[creds]`

`base.py`

Class: `StorageAdaptor`

Class: `TQDMHolder`

`irods_adaptor.py`

Class: `IRODSAdaptor`

`s3_adaptor.py`

Class: `S3Adaptor`

`fs_adaptor.py`

Class: `FSAdaptor`

`scp_adaptor.py`

Class: `SCPAdaptor`