Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def check_dataset_metadata_in_storage_unit(nco, dataset_dir):
assert len(nco.variables['dataset']) == 1 # 1 time slice
stored_metadata = nco.variables['dataset'][0]
if not isinstance(stored_metadata, str):
stored_metadata = netCDF4.chartostring(stored_metadata)
stored_metadata = str(np.char.decode(stored_metadata))
ds_filename = dataset_dir / 'agdc-metadata.yaml'
stored = yaml.safe_load(stored_metadata)
[(_, original)] = read_documents(ds_filename)
assert len(stored['lineage']['source_datasets']) == 1
assert next(iter(stored['lineage']['source_datasets'].values())) == original
def _test_read_docs_impl(sample_documents: Iterable[Tuple[str, int]]):
# Test case for returning URIs pointing to documents
for doc_url, num_docs in sample_documents:
all_docs = list(read_documents(doc_url, uri=True))
assert len(all_docs) == num_docs
for uri, doc in all_docs:
assert isinstance(doc, dict)
assert isinstance(uri, str)
url = as_url(doc_url)
if num_docs > 1:
expect_uris = [as_url(url) + '#part={}'.format(i) for i in range(num_docs)]
else:
expect_uris = [as_url(url)]
assert [f for f, _ in all_docs] == expect_uris
def load_dataset_definition(path):
if not isinstance(path, pathlib.Path):
path = pathlib.Path(path)
fname = get_metadata_path(path)
for _, doc in read_documents(fname):
return SimpleDocNav(doc)
def add_products(index, allow_exclusive_lock, files):
# type: (Index, bool, list) -> None
"""
Add or update products in the generic index.
"""
for descriptor_path, parsed_doc in read_documents(*files):
try:
type_ = index.products.from_doc(parsed_doc)
index.products.add(type_, allow_table_lock=allow_exclusive_lock)
echo('Added "%s"' % type_.name)
except InvalidDocException as e:
_LOG.exception(e)
_LOG.error('Invalid product definition: %s', descriptor_path)
sys.exit(1)
log('Checking indexes/views.')
index.metadata_types.check_field_indexes(
allow_table_lock=True,
rebuild_indexes=False,
rebuild_views=True,
)
if with_eo1:
log_header('Checking DEA eo1 metadata types')
# Add DEA metadata types, products.
for _, md_type_def in read_documents(DEA_MD_TYPES):
md = index.metadata_types.add(index.metadata_types.from_doc(md_type_def))
log(f"{md.name}")
log_header('Checking DEA products')
for _, product_def in read_documents(*DEA_PRODUCTS_DIR.glob('*.odc-product.yaml')):
product = index.products.add_document(product_def)
log(f"{product.name}")
log_header('Checking DEA ingested definitions')
for path in DEA_INGESTION_DIR.glob('*.yaml'):
ingest_config = ingest.load_config_from_file(path)
driver_name = ingest_config['storage']['driver']
driver = storage_writer_by_name(driver_name)
if driver is None:
raise ValueError("No driver found for {}".format(driver_name))
source_type, output_type = ingest.ensure_output_type(
index, ingest_config, driver.format, allow_product_changes=True
)
def _read_docs(paths):
return read_documents(*(Path(f) for f in paths))
def _path_dataset_ids(path: Path) -> Iterable[uuid.UUID]:
for _, metadata_doc in read_documents(path):
if metadata_doc is None:
raise InvalidDocException("Empty document from path {}".format(path))
if 'id' not in metadata_doc:
raise InvalidDocException("No id in path metadata: {}".format(path))
yield uuid.UUID(metadata_doc['id'])
def read_config(config_file):
_, stats_config = next(read_documents(config_file))
return stats_config
def _path_doc_stream(files, on_error, uri=True, raw=False):
"""See :func:`ui_path_doc_stream` for documentation"""
maybe_wrap = identity if raw else SimpleDocNav
for fname in files:
try:
for p, doc in read_documents(fname, uri=uri):
yield p, maybe_wrap(doc)
except InvalidDocException as e:
on_error(fname, e)