Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"""
if method == 'Get':
# full HTTP request url
request_url = build_get_url(url, data, overwrite=True)
log.debug(request_url)
# split URL into base url and query string to use utility function
spliturl = request_url.split('?')
u = openURL(spliturl[0], spliturl[
1], method='Get', username=username, password=password,
headers=headers, verify=verify, cert=cert, timeout=self.timeout)
return etree.fromstring(u.read())
elif method == 'Post':
u = openURL(url, data, method='Post',
username=username, password=password,
headers=headers, verify=verify, cert=cert, timeout=timeout)
return etree.fromstring(u.read())
else:
raise Exception("Unrecognized HTTP method: %s" % method)
def read(self, service_url):
"""
Get and parse a WMS capabilities document, returning an
elementtree instance
service_url is the base url, to which is appended the service,
acceptVersions, and request parameters
"""
getcaprequest = self.capabilities_url(service_url)
spliturl=getcaprequest.split('?')
u = openURL(spliturl[0], spliturl[1], method='Get', username=self.username, password=self.password)
return etree.fromstring(u.read())
url = self.reference
if url is None:
return ""
# a) 'http://cida.usgs.gov/climate/gdp/process/RetrieveResultServlet?id=1318528582026OUTPUT.601bb3d0-547f-4eab-8642-7c7d2834459e' # noqa
# b) 'http://rsg.pml.ac.uk/wps/wpsoutputs/outputImage-11294Bd6l2a.tif'
log.info('Output URL=%s' % url)
if '?' in url:
spliturl = url.split('?')
u = openURL(spliturl[0], spliturl[
1], method='Get', username=username, password=password,
headers=headers, verify=verify, cert=cert)
# extract output filepath from URL query string
self.fileName = spliturl[1].split('=')[1]
else:
u = openURL(
url, '', method='Get', username=username, password=password,
headers=headers, verify=verify, cert=cert)
# extract output filepath from base URL
self.fileName = url.split('/')[-1]
return u.read()
styles=styles,
srs=srs,
bbox=bbox,
format=format,
size=size,
time=time,
transparent=transparent,
bgcolor=bgcolor,
exceptions=exceptions,
**kwargs)
data = urlencode(request)
self.request = bind_url(base_url) + data
u = openURL(base_url, data, method, username=self.username, password=self.password, timeout=timeout or self.timeout)
# check for service exceptions, and return
if u.info()['Content-Type'].split(';')[0] in ['application/vnd.ogc.se_xml']:
se_xml = u.read()
se_tree = etree.fromstring(se_xml)
err_message = six.text_type(se_tree.find('ServiceException').text).strip()
raise ServiceException(err_message)
return u
def _gettilefromset(self, tilesets, x, y,z, ext, timeout=None):
for tileset in tilesets:
if tileset['order'] == z:
url = tileset['href'] + '/' + str(x) +'/' + str(y) + '.' + ext
u = openURL(url, '', username = self.username,
password = self.password, timeout=timeout or self.timeout)
return u
else:
raise ValueError('cannot find zoomlevel %i for TileMap' % z)
def parse_remote_metadata(self, timeout=30):
"""Parse remote metadata for MetadataURL of format 'text/xml' and add it as metadataUrl['metadata']"""
for metadataUrl in self.metadataUrls:
if metadataUrl['url'] is not None \
and metadataUrl['format'].lower() == 'text/xml':
try:
content = openURL(metadataUrl['url'], timeout=timeout)
doc = etree.fromstring(content.read())
if metadataUrl['type'] == 'FGDC':
mdelem = doc.find('.//metadata')
if mdelem is not None:
metadataUrl['metadata'] = Metadata(mdelem)
else:
metadataUrl['metadata'] = None
elif metadataUrl['type'] in ['TC211', '19115', '19139']:
mdelem = doc.find('.//'+util.nspath_eval('gmd:MD_Metadata', namespaces)) or doc.find('.//'+util.nspath_eval('gmi:MI_Metadata', namespaces))
if mdelem is not None:
metadataUrl['metadata'] = MD_Metadata(mdelem)
else:
metadataUrl['metadata'] = None
except:
metadataUrl['metadata'] = None
"value": value
}
}
}
if kwargs:
for kw in kwargs:
request[kw]=kwargs[kw]
#return request
data = json.dumps(request)
# print data
response = openURL(base_url, data, method, username=self.username, password=self.password).read()
return response
except StopIteration:
base_url = self.url
request = {'service': 'WFS', 'version': self.version, 'request': 'GetPropertyValue'}
if query:
request['query'] = str(query)
if valuereference:
request['valueReference'] = str(valuereference)
if storedquery_id:
request['storedQuery_id'] = str(storedquery_id)
if typename:
request['typename']=str(typename)
if kwargs:
for kw in kwargs.keys():
request[kw]=str(kwargs[kw])
encoded_request=urlencode(request)
u = openURL(base_url + encoded_request, timeout=self.timeout,
username=self.username, password=self.password)
return u.read()
request['resy']=resy
if resz:
request['resz']=resz
#anything else e.g. vendor specific parameters must go through kwargs
if kwargs:
for kw in kwargs:
request[kw]=kwargs[kw]
#encode and request
data = urlencode(request)
if log.isEnabledFor(logging.DEBUG):
log.debug('WCS 1.0.0 DEBUG: Second part of URL: %s'%data)
u=openURL(base_url, data, method, self.cookies)
return u
def parse_remote_metadata(self, timeout=30):
"""Parse remote metadata for MetadataURL and add it as metadataUrl['metadata']"""
for metadataUrl in self.metadataUrls:
if metadataUrl['url'] is not None:
try:
content = openURL(metadataUrl['url'], timeout=timeout)
doc = etree.fromstring(content.read())
mdelem = doc.find('.//metadata')
if mdelem is not None:
metadataUrl['metadata'] = Metadata(mdelem)
continue
mdelem = doc.find('.//' + util.nspath_eval('gmd:MD_Metadata', n.get_namespaces(['gmd']))) \
or doc.find('.//' + util.nspath_eval('gmi:MI_Metadata', n.get_namespaces(['gmi'])))
if mdelem is not None:
metadataUrl['metadata'] = MD_Metadata(mdelem)
continue
except:
metadataUrl['metadata'] = None