Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
- cql: common query language text. Note this overrides bbox, qtype, keywords
"""
# construct request
node0 = etree.Element(util.nspath_eval('csw:Transaction', namespaces))
node0.set('version', self.version)
node0.set('service', self.service)
node0.set(util.nspath_eval('xsi:schemaLocation', namespaces), schema_location)
validtransactions = ['insert', 'update', 'delete']
if ttype not in validtransactions: # invalid transaction
raise RuntimeError, 'Invalid transaction \'%s\'.' % ttype
node1 = etree.SubElement(node0, util.nspath_eval('csw:%s' % ttype.capitalize(), namespaces))
if ttype != 'update':
node1.set('typeName', typename)
if ttype == 'insert':
if record is None:
raise RuntimeError, 'Nothing to insert.'
node1.append(etree.fromstring(record))
if ttype == 'update':
if record is not None:
node1.append(etree.fromstring(record))
else:
if propertyname is not None and propertyvalue is not None:
node2 = etree.SubElement(node1, util.nspath_eval('csw:RecordProperty', namespaces))
etree.SubElement(node2, util.nspath_eval('csw:Name', namespaces)).text = propertyname
def _setpropertyisequalto(self, parent, propertyname, literal, matchcase=True):
"""
construct a PropertyIsEqualTo
Parameters
----------
- parent: parent etree.Element object
- propertyname: the PropertyName
- literal: the Literal value
- matchcase: whether to perform a case insensitve query (default is True)
"""
tmp = etree.SubElement(parent, util.nspath_eval('ogc:PropertyIsEqualTo', namespaces))
if matchcase is False:
tmp.set('matchCase', 'false')
etree.SubElement(tmp, util.nspath_eval('ogc:PropertyName', namespaces)).text = propertyname
etree.SubElement(tmp, util.nspath_eval('ogc:Literal', namespaces)).text = literal
node0.set('outputFormat', format)
node0.set('version', self.version)
node0.set('service', self.service)
node0.set('resultType', resulttype)
if startposition > 0:
node0.set('startPosition', str(startposition))
node0.set('maxRecords', str(maxrecords))
node0.set(util.nspath_eval('xsi:schemaLocation', namespaces), schema_location)
node1 = etree.SubElement(node0, util.nspath_eval('csw:Query', namespaces))
node1.set('typeNames', typenames)
etree.SubElement(node1, util.nspath_eval('csw:ElementSetName', namespaces)).text = esn
if any([len(constraints) > 0, cql is not None]):
node2 = etree.SubElement(node1, util.nspath_eval('csw:Constraint', namespaces))
node2.set('version', '1.1.0')
flt = fes.FilterRequest()
if len(constraints) > 0:
node2.append(flt.setConstraintList(constraints))
# Now add a CQL filter if passed in
elif cql is not None:
etree.SubElement(node2, util.nspath_eval('csw:CqlText', namespaces)).text = cql
if sortby is not None and isinstance(sortby, fes.SortBy):
node1.append(sortby.toXML())
self.request = node0
self._invoke()
if self.exceptionreport is None:
def toXML(self):
node0 = etree.Element(util.nspath_eval('ogc:PropertyIsBetween', namespaces))
etree.SubElement(node0, util.nspath_eval('ogc:PropertyName', namespaces)).text = self.propertyname
node1 = etree.SubElement(node0, util.nspath_eval('ogc:LowerBoundary', namespaces))
etree.SubElement(node1, util.nspath_eval('ogc:Literal', namespaces)).text = '%s' % self.lower
node2 = etree.SubElement(node0, util.nspath_eval('ogc:UpperBoundary', namespaces))
etree.SubElement(node2, util.nspath_eval('ogc:Literal', namespaces)).text = '%s' % self.upper
return node0
encodes base OwcContext as dict to atom xml tree
:param d:
:return:
"""
xml = etree.Element("feed", nsmap=ns)
etree.SubElement(xml, "id").text = d['id']
spec_reference = [axml_link(do) for do in
extract_p('properties.links.profiles', d, [])]
[xml.append(el) for el in spec_reference if el is not None]
area_of_interest = extract_p('bbox', d, None)
if area_of_interest is not None:
try:
gml = etree.fromstring(area_of_interest)
georss = etree.SubElement(xml, ns_elem("georss", "where"))
georss.append(gml)
except Exception as ex:
log.warn('could encode bbox into georss:where', ex)
pass
context_metadata = [axml_link(do) for do in
extract_p('properties.links.via', d, [])]
[xml.append(el) for el in context_metadata if el is not None]
language = extract_p('properties.lang', d, None)
if language is not None: xml.set(ns_elem("xml", "lang"), language)
title = extract_p('properties.title', d, None)
if title is not None: etree.SubElement(xml, "title").text = title
#
def setsortby(parent, propertyname, order='ASC'):
"""
constructs a SortBy element
Parameters
----------
- parent: parent etree.Element object
- propertyname: the PropertyName
- order: the SortOrder (default is 'ASC')
"""
tmp = etree.SubElement(parent, util.nspath_eval('ogc:SortBy', namespaces))
tmp2 = etree.SubElement(tmp, util.nspath_eval('ogc:SortProperty', namespaces))
etree.SubElement(tmp2, util.nspath_eval('ogc:PropertyName', namespaces)).text = propertyname
etree.SubElement(tmp2, util.nspath_eval('ogc:SortOrder', namespaces)).text = order
dataElement = etree.Element(nspath_eval('wps:Data', namespaces))
complexDataElement = etree.SubElement(
dataElement, nspath_eval('wps:ComplexData', namespaces),
attrib={"mimeType": "text/xml", "schema": GML_SCHEMA_LOCATION})
featureMembersElement = etree.SubElement(
complexDataElement, nspath_eval('gml:featureMembers', namespaces),
attrib={nspath_eval("xsi:schemaLocation", namespaces): "%s %s" % (DRAW_NAMESPACE, DRAW_SCHEMA_LOCATION)})
boxElement = etree.SubElement(featureMembersElement, nspath_eval(
'gml:box', namespaces), attrib={nspath_eval("gml:id", namespaces): "box.1"})
geomElement = etree.SubElement(
boxElement, nspath_eval('gml:the_geom', namespaces))
multiPolygonElement = etree.SubElement(
geomElement, nspath_eval('gml:MultiPolygon', namespaces),
attrib={"srsDimension": "2", "srsName": "http://www.opengis.net/gml/srs/epsg.xml#4326"})
for polygon in self.polygons:
polygonMemberElement = etree.SubElement(
multiPolygonElement, nspath_eval('gml:polygonMember', namespaces))
polygonElement = etree.SubElement(
polygonMemberElement, nspath_eval('gml:Polygon', namespaces))
exteriorElement = etree.SubElement(
polygonElement, nspath_eval('gml:exterior', namespaces))
linearRingElement = etree.SubElement(
exteriorElement, nspath_eval('gml:LinearRing', namespaces))
posListElement = etree.SubElement(
linearRingElement, nspath_eval('gml:posList', namespaces))
posListElement.text = ' '.join(
["%s %s" % (x, y) for x, y in polygon[:]])
idElement = etree.SubElement(
boxElement, nspath_eval('gml:ID', namespaces))
idElement.text = "0"
return dataElement
#
# the_geom
# STATE
#
#
#
#
queryElement = etree.Element(
nspath_eval('wfs:Query', namespaces), attrib={"typeName": self.typeName})
for propertyName in self.propertyNames:
propertyNameElement = etree.SubElement(
queryElement, nspath_eval('wfs:PropertyName', namespaces))
propertyNameElement.text = propertyName
if len(self.filters) > 0:
filterElement = etree.SubElement(
queryElement, nspath_eval('ogc:Filter', namespaces))
for filter in self.filters:
# gmlObjectIdElement
etree.SubElement(
filterElement, nspath_eval('ogc:GmlObjectId', namespaces),
attrib={nspath_eval('gml:id', namespaces): filter})
return queryElement
if etree.__name__ != 'lxml.etree': # apply nsmap manually
node0.set('xmlns:ows', namespaces['ows'])
node0.set('xmlns:gmd', namespaces['gmd'])
node0.set('xmlns:dif', namespaces['dif'])
node0.set('xmlns:fgdc', namespaces['fgdc'])
node0.set('outputSchema', outputschema)
node0.set('outputFormat', format)
node0.set('version', self.version)
node0.set('resultType', resulttype)
node0.set('service', self.service)
if startposition > 0:
node0.set('startPosition', str(startposition))
node0.set('maxRecords', str(maxrecords))
node0.set(util.nspath_eval('xsi:schemaLocation', namespaces), schema_location)
node1 = etree.SubElement(node0, util.nspath_eval('csw:Query', namespaces))
node1.set('typeNames', typenames)
etree.SubElement(node1, util.nspath_eval('csw:ElementSetName', namespaces)).text = esn
self._setconstraint(node1, qtype, propertyname, keywords, bbox, cql, None)
if sortby is not None:
fes.setsortby(node1, sortby)
self.request = node0
self._invoke()
if self.exceptionreport is None:
self.results = {}
attrib={"srsDimension": "2", "srsName": "http://www.opengis.net/gml/srs/epsg.xml#4326"})
for polygon in self.polygons:
polygonMemberElement = etree.SubElement(
multiPolygonElement, nspath_eval('gml:polygonMember', namespaces))
polygonElement = etree.SubElement(
polygonMemberElement, nspath_eval('gml:Polygon', namespaces))
exteriorElement = etree.SubElement(
polygonElement, nspath_eval('gml:exterior', namespaces))
linearRingElement = etree.SubElement(
exteriorElement, nspath_eval('gml:LinearRing', namespaces))
posListElement = etree.SubElement(
linearRingElement, nspath_eval('gml:posList', namespaces))
posListElement.text = ' '.join(
["%s %s" % (x, y) for x, y in polygon[:]])
idElement = etree.SubElement(
boxElement, nspath_eval('gml:ID', namespaces))
idElement.text = "0"
return dataElement