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_csw_ngdc():
"rewritten doctest/cws_ngdc.txt"
c = csw.CatalogueServiceWeb(SERVICE_URL, timeout=120)
assert c.identification.title == 'ArcGIS Server Geoportal Extension 10 - OGC CSW 2.0.2 ISO AP'
assert c.identification.version == '2.0.2'
assert sorted(c.identification.keywords) == [
'Geophysical Metadata', 'NGDC', 'Ocean Metadata', 'Space Weather Metadata']
assert c.provider.name == 'NOAA NGDC'
# Get some records
sos_urn = 'urn:x-esri:specification:ServiceType:sos:url'
aoos_uuid = '1706F520-2647-4A33-B7BF-592FAFDE4B45'
uuid_filter = fes.PropertyIsEqualTo(propertyname='sys.siteuuid', literal="{%s}" % aoos_uuid)
c.getrecords2([uuid_filter], esn='full', maxrecords=999999)
assert len(c.records) > 40
assert 'AOOS SOS' in c.records
'distributedsearch' in self.parent.kvp and
self.parent.kvp['distributedsearch'] and self.parent.kvp['hopcount'] > 0):
# do distributed search
LOGGER.debug('DistributedSearch specified (hopCount: %s)',
self.parent.kvp['hopcount'])
from owslib.csw import CatalogueServiceWeb
from owslib.ows import ExceptionReport
for fedcat in \
self.parent.config.get('server', 'federatedcatalogues').split(','):
LOGGER.info('Performing distributed search on federated \
catalogue: %s', fedcat)
try:
start_time = time()
remotecsw = CatalogueServiceWeb(fedcat, skip_caps=True)
remotecsw.getrecords2(xml=self.parent.request,
esn=self.parent.kvp['elementsetname'],
outputschema=self.parent.kvp['outputschema'])
fsr = etree.SubElement(searchresults, util.nspath_eval(
'csw30:FederatedSearchResult',
self.parent.context.namespaces),
catalogueURL=fedcat.request)
msg = 'Distributed search results from catalogue %s: %s.' % (fedcat, remotecsw.results)
LOGGER.debug(msg)
fsr.append(etree.Comment(msg))
search_result = etree.SubElement(fsr, util.nspath_eval(
'csw30:searchResult', self.parent.context.namespaces),
recordSchema=self.parent.kvp['outputschema'],
# simple process to harvest CSW catalogues via Harvest operations
import sys
from owslib.csw import CatalogueServiceWeb
stop = 0
flag = 0
maxrecords = 10
if len(sys.argv) < 3:
print('Usage: %s [maxrecords]' \
% sys.argv[0])
sys.exit(1)
src = CatalogueServiceWeb(sys.argv[1])
dest = CatalogueServiceWeb(sys.argv[2])
if len(sys.argv) == 4:
maxrecords = sys.argv[3]
while stop == 0:
if flag == 0: # first run, start from 0
startposition = 0
else: # subsequent run, startposition is now paged
startposition = src.results['nextrecord']
src.getrecords(esn='brief', startposition=startposition, maxrecords=maxrecords)
print(src.results)
if src.results['nextrecord'] == 0 \
or src.results['returned'] == 0 \
def harvest(source, dst):
maxrecords = options['max']
if options['max'] == 0 or None:
maxrecords = 10
stop = 0
flag = 0
src = CatalogueServiceWeb(source)
dest = CatalogueServiceWeb(dst)
while stop == 0:
if flag == 0: # first run, start from 0
startposition = 0
else: # subsequent run, startposition is now paged
startposition = src.results['nextrecord']
src.getrecords(esn='brief', startposition=startposition, maxrecords=maxrecords)
print(src.results)
if src.results['nextrecord'] == 0 \
or src.results['returned'] == 0 \
or src.results['nextrecord'] > src.results['matches']: # end the loop, exhausted all records
stop = 1
def _parse_csw(context, repos, record, identifier, pagesize=10):
from owslib.csw import CatalogueServiceWeb
recobjs = [] # records
serviceobj = repos.dataset()
# if init raises error, this might not be a CSW
md = CatalogueServiceWeb(record, timeout=60)
LOGGER.info('Setting CSW service metadata')
# generate record of service instance
_set(context, serviceobj, 'pycsw:Identifier', identifier)
_set(context, serviceobj, 'pycsw:Typename', 'csw:Record')
_set(context, serviceobj, 'pycsw:Schema', 'http://www.opengis.net/cat/csw/2.0.2')
_set(context, serviceobj, 'pycsw:MdSource', record)
_set(context, serviceobj, 'pycsw:InsertDate', util.get_today_and_now())
_set(
context,
serviceobj,
'pycsw:AnyText',
util.get_anytext(md._exml)
)
_set(context, serviceobj, 'pycsw:Type', 'service')
_set(context, serviceobj, 'pycsw:Title', md.identification.title)
endpoint = domain
detected = True
if domain in [
'http://maps.nypl.org/',
'http://mapwarper.net/',
'http://warp.worldmap.harvard.edu/',
]:
service_type = 'Hypermap:WARPER'
title = 'Warper at %s' % domain
abstract = 'Warper at %s' % domain
detected = True
# test if it is CSW, WMS, TMS, WMTS or Esri
# CSW
try:
csw = CatalogueServiceWeb(endpoint)
service_type = 'OGC:CSW'
service_links = {}
detected = True
typenames = 'csw:Record'
outputschema = 'http://www.opengis.net/cat/csw/2.0.2'
if 'csw_harvest_pagesize' in settings.REGISTRY_PYCSW['manager']:
pagesize = int(settings.REGISTRY_PYCSW['manager']['csw_harvest_pagesize'])
else:
pagesize = 10
LOGGER.debug('Harvesting CSW %s' % endpoint)
# now get all records
# get total number of records to loop against
try:
def _init_service(self):
if self._catalogue:
return
if not self._catalogue_service:
self._catalogue_service = CatalogueServiceWeb(url=self._catalogue_url, timeout=_CSW_TIMEOUT, skip_caps=True)
GWarning('Error getting response: %s' % err)
return
except KeyError, err:
GWarning('Record parsing error, unable to locate record identifier')
return
if self.catalog:
md = cat.records.values()[0]
path = 'record_metadata_gmd.html'
metadata = render_template('en', self.context, md, path)
self.htmlView.SetPage(metadata)
else:
identifier = self.get_item_data(idx, 'identifier')
cat = CatalogueServiceWeb(self.catalog_url, timeout=self.timeout)
try:
cat.getrecordbyid([self.catalog.records[identifier].identifier])
except ExceptionReport, err:
GWarning('Error getting response: %s' % err)
return
except KeyError, err:
GWarning('Record parsing error, unable to locate record identifier')
return
record = cat.records[identifier]
record.xml_url = cat.request
if self.catalog:
path = 'record_metadata_dc.html'
metadata = render_template('en', self.context,
record,
service = Service.objects.get(base_url=url)
return_dict = [{
'status': 'ok',
'service_id': service.pk,
'service_name': service.name,
'service_title': service.title,
'msg': 'This is an existing Service'
}]
return HttpResponse(json.dumps(return_dict),
mimetype='application/json',
status=200)
except:
pass
if csw is None:
csw = CatalogueServiceWeb(url)
service = Service.objects.create(base_url=url,
type='CSW',
method='H',
name=_get_valid_name(
csw.identification.title or url),
title=csw.identification.title,
version=csw.identification.version,
abstract=csw.identification.abstract or _("Not provided"),
owner=owner)
service.keywords = ','.join(csw.identification.keywords)
service.save
service.set_default_permissions()
message = "Service %s registered" % service.name