pass kwargs and allow env options
@@ -25,12 +25,15 @@ def _get_url_size(fname, **open_kwargs):
return size
-def _copy_btw_filesystems(input_opener, output_opener, BLOCK_SIZE=10_000_000):
+def _copy_btw_filesystems(input_opener, output_opener, BLOCK_SIZE=10_000_000, **kwargs):
+ streaming = True if kwargs.pop("block_size", True) == 0 else False
+ if os.getenv("PANGEO_FORGE_BLOCK_SIZE"):
+ BLOCK_SIZE = int(os.getenv("PANGEO_FORGE_BLOCK_SIZE"))
with input_opener as source:
with output_opener as target:
count = summed_bytes = 0
while True:
- logger.debug("_copy_btw_filesystems reading") if source.block_size != 0 else None
+ logger.debug("_copy_btw_filesystems reading data") if not streaming else None
try:
data = source.read(BLOCK_SIZE)
except BlockSizeError as e:
@@ -41,15 +44,13 @@ def _copy_btw_filesystems(input_opener, output_opener, BLOCK_SIZE=10_000_000):
if not data:
break
# strided logging pattern for streaming transfers to avoid excessive logs
- if source.block_size == 0:
+ if streaming:
summed_bytes += len(data)
if summed_bytes // BLOCK_SIZE >= count:
logger.debug(
f"_copy_btw_filesystems copying block {count} of ~{BLOCK_SIZE} bytes"
)
count += 1
- else:
- pass
else:
logger.debug(f"_copy_btw_filesystems copying block of {len(data)} bytes")
target.write(data)
@@ -162,7 +163,7 @@ class CacheFSSpecTarget(FlatFSSpecTarget):
input_opener = fsspec.open(fname, mode="rb", **open_kwargs)
target_opener = self.open(fname, mode="wb")
logger.info(f"Coping remote file '{fname}' to cache")
- _copy_btw_filesystems(input_opener, target_opener)
+ _copy_btw_filesystems(input_opener, target_opener, **open_kwargs)
class MetadataTarget(FSSpecTarget):
