Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_path_or_store_write_to_bucket(self):
path, _, _ = _get_path_or_store('http://obs.eu-de.otc.t-systems.com/fake_bucket/fake_cube.zarr',
mode='write',
client_kwargs={'aws_access_key_id': 'some_fake_id',
'aws_secret_access_key': 'some_fake_key'})
self.assertIsInstance(path, fsspec.mapping.FSMap)
def test_path_or_store_read_from_bucket(self):
path, _, _ = _get_path_or_store('http://obs.eu-de.otc.t-systems.com/dcs4cop-obs-02/OLCI-SNS-RAW-CUBE-2.zarr',
mode='read')
self.assertIsInstance(path, fsspec.mapping.FSMap)
url: str
Root URL of mapping
check: bool
Whether to attempt to read from the location before instantiation, to
check that the mapping does exist
create: bool
Whether to make the directory corresponding to the root before
instantiating
Returns
-------
``FSMap`` instance, the dict-like key-value store.
"""
# Removing protocol here - could defer to each open() on the backend
fs, urlpath = url_to_fs(url, **kwargs)
return FSMap(urlpath, fs, check, create)
def get_mapper(self, root, check=False, create=False):
"""Create key/value store based on this file-system
Makes a MutibleMapping interface to the FS at the given root path.
See ``fsspec.mapping.FSMap`` for further details.
"""
from .mapping import FSMap
return FSMap(root, self, check, create)
def _open_asset(path, data_format, zarr_kwargs, cdf_kwargs, preprocess, varname):
protocol = None
root = path
if isinstance(path, fsspec.mapping.FSMap):
protocol = path.fs.protocol
if isinstance(protocol, list):
protocol = tuple(protocol)
if protocol in {'http', 'https', 'file'} or protocol is None:
path = path.root
root = path
else:
root = path.root
if data_format == 'zarr':
logger.debug(f'Opening zarr store: {root} - protocol: {protocol}')
try:
ds = xr.open_zarr(path, **zarr_kwargs)
except Exception as e: