Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"""Same as siphon.catalog.TDSCatalog, but with constructor which takes url and response
So it can be used by a generic asynchronous web crawler
Args:
catalog_url: URL with THREDDS catalog xml file
content: Content found at catalog_url
"""
self.catalog_url = catalog_url
self.base_tds_url = _find_base_tds_url(self.catalog_url)
# begin parsing the xml doc
root = ET.fromstring(content)
self.catalog_name = root.attrib.get('name', 'No name found')
self.datasets = DatasetCollection()
self.services = []
self.catalog_refs = DatasetCollection()
self.metadata = {}
self.ds_with_access_elements_to_process = []
service_skip_count = 0
service_skip = 0
current_dataset = None
previous_dataset = None
for child in root.iter():
tag_type = child.tag.split('}')[-1]
if tag_type == 'dataset':
current_dataset = child.attrib['name']
self._process_dataset(child)
# see if the previously processed dataset has access elements as children
# if so, these datasets need to be processed specially when making
So it can be used by a generic asynchronous web crawler
Args:
catalog_url: URL with THREDDS catalog xml file
content: Content found at catalog_url
"""
self.catalog_url = catalog_url
self.base_tds_url = _find_base_tds_url(self.catalog_url)
# begin parsing the xml doc
root = ET.fromstring(content)
self.catalog_name = root.attrib.get('name', 'No name found')
self.datasets = DatasetCollection()
self.services = []
self.catalog_refs = DatasetCollection()
self.metadata = {}
self.ds_with_access_elements_to_process = []
service_skip_count = 0
service_skip = 0
current_dataset = None
previous_dataset = None
for child in root.iter():
tag_type = child.tag.split('}')[-1]
if tag_type == 'dataset':
current_dataset = child.attrib['name']
self._process_dataset(child)
# see if the previously processed dataset has access elements as children
# if so, these datasets need to be processed specially when making
# access_urls
if previous_dataset and self.datasets[previous_dataset].access_element_info:
# If we were given an HTML link, warn about it and try to fix to xml
if 'html' in resp.headers['content-type']:
import warnings
new_url = self.catalog_url.replace('html', 'xml')
warnings.warn('URL {} returned HTML. Changing to: {}'.format(self.catalog_url,
new_url))
self.catalog_url = new_url
resp = self.session.get(self.catalog_url)
resp.raise_for_status()
# begin parsing the xml doc
root = ET.fromstring(resp.content)
self.catalog_name = root.attrib.get('name', 'No name found')
self.datasets = DatasetCollection()
self.services = []
self.catalog_refs = DatasetCollection()
self.metadata = {}
self.ds_with_access_elements_to_process = []
service_skip_count = 0
service_skip = 0
current_dataset = None
previous_dataset = None
for child in root.iter():
tag_type = child.tag.split('}')[-1]
if tag_type == 'dataset':
current_dataset = child.attrib['name']
self._process_dataset(child)
if previous_dataset:
# see if the previously processed dataset has access elements as children
if 'html' in resp.headers['content-type']:
import warnings
new_url = self.catalog_url.replace('html', 'xml')
warnings.warn('URL {} returned HTML. Changing to: {}'.format(self.catalog_url,
new_url))
self.catalog_url = new_url
resp = self.session.get(self.catalog_url)
resp.raise_for_status()
# begin parsing the xml doc
root = ET.fromstring(resp.content)
self.catalog_name = root.attrib.get('name', 'No name found')
self.datasets = DatasetCollection()
self.services = []
self.catalog_refs = DatasetCollection()
self.metadata = {}
self.ds_with_access_elements_to_process = []
service_skip_count = 0
service_skip = 0
current_dataset = None
previous_dataset = None
for child in root.iter():
tag_type = child.tag.split('}')[-1]
if tag_type == 'dataset':
current_dataset = child.attrib['name']
self._process_dataset(child)
if previous_dataset:
# see if the previously processed dataset has access elements as children
# if so, these datasets need to be processed specially when making
# access_urls