Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _write_codelist_element(codelist_element, codelist_value, nsmap):
namespace, codelist = codelist_element.split(':')
element = etree.Element(util.nspath_eval(codelist_element, nsmap),
codeSpace=CODESPACE, codeList='%s#%s' % (CODELIST, codelist),
codeListValue=codelist_value)
element.text = codelist_value
return element
keywords = util.getqattr(recobj, queryables['dc:subject']['dbcol'])
if keywords is not None:
for keyword in keywords.split(','):
etree.SubElement(record,
util.nspath_eval('dc:subject',
self.parent.context.namespaces)).text = keyword
val = util.getqattr(recobj, self.parent.context.md_core_model['mappings']['pycsw:TopicCategory'])
if val:
etree.SubElement(record,
util.nspath_eval('dc:subject',
self.parent.context.namespaces), scheme='http://www.isotc211.org/2005/resources/Codelist/gmxCodelists.xml#MD_TopicCategoryCode').text = val
val = util.getqattr(recobj, queryables['dc:format']['dbcol'])
if val:
etree.SubElement(record,
util.nspath_eval('dc:format',
self.parent.context.namespaces)).text = val
# links
rlinks = util.getqattr(recobj,
self.parent.context.md_core_model['mappings']['pycsw:Links'])
if rlinks:
links = rlinks.split('^')
for link in links:
linkset = link.split(',')
etree.SubElement(record,
util.nspath_eval('dct:references',
self.parent.context.namespaces),
scheme=linkset[2]).text = linkset[-1]
# bbox extent
val = util.getqattr(result, queryables['apiso:BoundingBox']['dbcol'])
bboxel = write_extent(val, self.namespaces)
if bboxel is not None and mtype != 'service':
resident.append(bboxel)
# service identification
if mtype == 'service':
# service type
# service type version
val = util.getqattr(result, queryables['apiso:ServiceType']['dbcol'])
val2 = util.getqattr(result, queryables['apiso:ServiceTypeVersion']['dbcol'])
if val is not None:
tmp = etree.SubElement(resident, util.nspath_eval('srv:serviceType', self.namespaces))
etree.SubElement(tmp, util.nspath_eval('gco:LocalName', self.namespaces)).text = val
tmp = etree.SubElement(resident, util.nspath_eval('srv:serviceTypeVersion', self.namespaces))
etree.SubElement(tmp, util.nspath_eval('gco:CharacterString', self.namespaces)).text = val2
kw = util.getqattr(result, queryables['apiso:Subject']['dbcol'])
if kw is not None:
srv_keywords = etree.SubElement(resident, util.nspath_eval('srv:keywords', self.namespaces))
srv_keywords.append(write_keywords(kw, self.namespaces))
if bboxel is not None:
bboxel.tag = util.nspath_eval('srv:extent', self.namespaces)
resident.append(bboxel)
val = util.getqattr(result, queryables['apiso:CouplingType']['dbcol'])
if val is not None:
couplingtype = etree.SubElement(resident, util.nspath_eval('srv:couplingType', self.namespaces))
etree.SubElement(couplingtype, util.nspath_eval('srv:SV_CouplingType', self.namespaces), codeListValue=val, codeList='%s#SV_CouplingType' % CODELIST).text = val
elif (par_count > 1):
LOGGER.debug('ogc:And query (%d predicates)', par_count)
# Since more than 1 parameter, append the AND logical operator
logical_and = etree.Element(util.nspath_eval('ogc:And',
context.namespaces))
if bbox_element is not None:
logical_and.append(bbox_element)
if time_element is not None:
logical_and.append(time_element)
if anytext_elements is not None:
logical_and.extend(anytext_elements)
root.append(logical_and)
# Render etree to string XML
LOGGER.debug(etree.tostring(root, encoding='unicode'))
return etree.tostring(root, encoding='unicode')