3903887078

Build Type

push

github

Committed by GitHub

Commit Message

Merge pull request #808 from envisionz/wfs-keywords

Run Details

7494 of 12703 relevant lines covered (58.99%)

0.59 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.41

/owslib/feature/postrequest.py

# owslib imports:
from owslib import util
from owslib.etree import etree
from owslib.namespaces import Namespaces

n = Namespaces()
FES_NAMESPACE = n.get_namespace("fes")
GML_NAMESPACE = n.get_namespace("gml")
GML32_NAMESPACE = n.get_namespace("gml32")
OGC_NAMESPACE = n.get_namespace("ogc")
WFS_NAMESPACE = n.get_namespace("wfs")
WFS20_NAMESPACE = n.get_namespace("wfs20")


class PostRequest():
    """Superclass for POST request building"""

    def __init__(self, version=None, namespace=None):
        self._root = etree.Element(util.nspath('GetFeature', namespace))
        self._root.set("service", "WFS")
        self._root.set("version", version)
        self._wfsnamespace = namespace
        self._query = None

    def _create_query(self, typename):
        self._query = etree.SubElement(self._root, util.nspath('Query', self._wfsnamespace))

    def set_featureversion(self, version):
        self._query.set("featureVersion", version)

    def set_propertyname(self, propertyname):
        """Set which feature properties will be returned.

        If not set, will return all properties."""
        for pn in propertyname:
            etree.SubElement(self._query, "PropertyName").text = pn

    def set_startindex(self, startindex):
        """Set the starting index value for the request"""
        self._root.set("startIndex", str(startindex))

    def to_string(self):
        """Returns the xml request in string format.

        Required in order to use the request with getfeature()"""
        return etree.tostring(self._root)


class PostRequest_1_1_0(PostRequest):
    """XML Post request payload builder for WFS version 1.1.0"""

    def __init__(self):
        super().__init__(version='1.1.0', namespace=WFS_NAMESPACE)

    def create_query(self, typename):
        """Creates the query tag with the corresponding typenames.
        Required element for each request."""
        super()._create_query(typename)
        self._query.set("typeName", typename)

    def set_bbox(self, bbox):
        """Set a bbox filter.

        Cannot be used with set_featureid() or set_filter().
        """
        filter_tree = etree.SubElement(self._query, util.nspath('Filter', OGC_NAMESPACE))
        bbox_tree = etree.SubElement(filter_tree, util.nspath('BBOX', OGC_NAMESPACE))
        coords = etree.SubElement(bbox_tree, util.nspath('Envelope', GML_NAMESPACE))
        if len(bbox) > 4:
            coords.set('srsName', bbox[4])
        lower = etree.SubElement(coords, util.nspath('lowerCorner', GML_NAMESPACE))
        lower.text = '{} {}'.format(bbox[0], bbox[1])

        upper = etree.SubElement(coords, util.nspath('upperCorner', GML_NAMESPACE))
        upper.text = '{} {}'.format(bbox[2], bbox[3])

    def set_featureid(self, featureid):
        """Set filter by feature id.

        Cannot be used with set_bbox() or set_filter().
        """
        feature_tree = etree.SubElement(self._query, util.nspath('Filter', OGC_NAMESPACE))

        for ft in featureid:
            prop_id = etree.Element(util.nspath('GmlObjectId', OGC_NAMESPACE))
            prop_id.set(util.nspath('id', GML_NAMESPACE), ft)
            feature_tree.append(prop_id)

    def set_filter(self, filter):
        """Set filter from existing filter.

        Will integrate the filter tag of a provided xml filter to the query being built.

        Cannot be used with set_bbox() or set_featureid().
        """
        f = etree.fromstring(filter)
        sub_elem = f.find(util.nspath("Filter", OGC_NAMESPACE))
        self._query.append(sub_elem)

    def set_maxfeatures(self, maxfeatures):
        """Set the maximum number of features to be returned."""
        self._root.set("maxFeatures", str(maxfeatures))

    def set_outputformat(self, outputFormat):
        """Set the output format.

        Verify the available formats with a GetCapabilites request."""
        self._root.set("outputFormat", outputFormat)

    def set_sortby(self, sortby):
        """Set the properties by which the response will be sorted."""
        sort_tree = etree.SubElement(self._query, util.nspath("SortBy", OGC_NAMESPACE))
        for s in sortby:
            prop_elem = etree.SubElement(sort_tree, util.nspath("SortProperty", OGC_NAMESPACE))
            prop_name = etree.SubElement(prop_elem, util.nspath('PropertyName', OGC_NAMESPACE))
            prop_name.text = s


class PostRequest_2_0_0(PostRequest):
    """XML Post request payload builder for WFS version 2.0.0."""

    def __init__(self):
        super().__init__(version='2.0.0', namespace=WFS20_NAMESPACE)

    def create_query(self, typename):
        """Creates the query tag with the corresponding typenames.
        Required element for each request ecept for stored queries."""
        super()._create_query(typename)
        self._query.set("typenames", typename)

    def create_storedquery(self, stored_id, parameters):
        """Create the storedQuery tag and configure it's sub elements and attributes."""
        storedquery = etree.SubElement(self._root, util.nspath('StoredQuery', self._wfsnamespace))
        storedquery.set("id", str(stored_id))
        for param in parameters:
            p = etree.SubElement(storedquery, util.nspath('Parameter', self._wfsnamespace))
            p.set("name", param)
            p.text = parameters[param]

    def set_bbox(self, bbox):
        """Set a bbox filter.

        Cannot be used with set_featureid() or set_filter().
        """
        filter_tree = etree.SubElement(self._query, util.nspath('Filter', FES_NAMESPACE))
        bbox_tree = etree.SubElement(filter_tree, util.nspath('BBOX', FES_NAMESPACE))
        coords = etree.SubElement(bbox_tree, util.nspath('Envelope', GML32_NAMESPACE))
        if len(bbox) > 4:
            coords.set('srsName', bbox[4])

        lower = etree.SubElement(coords, util.nspath('lowerCorner', GML32_NAMESPACE))
        lower.text = '{} {}'.format(bbox[0], bbox[1])

        upper = etree.SubElement(coords, util.nspath('upperCorner', GML32_NAMESPACE))
        upper.text = '{} {}'.format(bbox[2], bbox[3])

    def set_featureid(self, featureid):
        """Set filter by feature id.

        Cannot be used with set_bbox() or set_filter().
        """
        feature_tree = etree.SubElement(self._query, util.nspath('Filter', FES_NAMESPACE))
        for ft in featureid:
            prop_id = etree.Element(util.nspath('ResourceId', FES_NAMESPACE))
            prop_id.set('rid', ft)
            feature_tree.append(prop_id)

    def set_filter(self, filter):
        """Set filter from existing filter.

        Will integrate the filter tag of a provided xml filter to the current one
        being built.

        Cannot be used with set_bbox() or set_featureid().
        """
        if isinstance(filter, str):
            f = etree.fromstring(filter)
            # If this raises an error, it means the filter string does not contain an actual Filter node.
            sub_elem = f.xpath("//FES:Filter", namespaces={"FES": FES_NAMESPACE})[0]

        else:
            sub_elem = filter

        self._query.append(sub_elem)

    def set_maxfeatures(self, maxfeatures):
        """Set the maximum number of features to be returned."""
        self._root.set("count", str(maxfeatures))

    def set_outputformat(self, outputFormat):
        """Set the output format.

        Verify the available formats with a GetCapabilites request.
        """
        self._root.set("outputformat", outputFormat)

    def set_sortby(self, sortby):
        """Set the properties by which the response will be sorted."""
        sort_tree = etree.SubElement(self._query, util.nspath("SortBy", FES_NAMESPACE))
        for s in sortby:
            prop_elem = etree.SubElement(sort_tree, util.nspath("SortProperty", FES_NAMESPACE))
            value = etree.SubElement(prop_elem, util.nspath('ValueReference', FES_NAMESPACE))
            value.text = s

1	# owslib imports:
2	from owslib import util	1✔
3	from owslib.etree import etree	1✔
4	from owslib.namespaces import Namespaces	1✔
5
6	n = Namespaces()	1✔
7	FES_NAMESPACE = n.get_namespace("fes")	1✔
8	GML_NAMESPACE = n.get_namespace("gml")	1✔
9	GML32_NAMESPACE = n.get_namespace("gml32")	1✔
10	OGC_NAMESPACE = n.get_namespace("ogc")	1✔
11	WFS_NAMESPACE = n.get_namespace("wfs")	1✔
12	WFS20_NAMESPACE = n.get_namespace("wfs20")	1✔
13
14
15	class PostRequest():	1✔
16	"""Superclass for POST request building"""
17
18	def __init__(self, version=None, namespace=None):	1✔
19	self._root = etree.Element(util.nspath('GetFeature', namespace))	1✔
20	self._root.set("service", "WFS")	1✔
21	self._root.set("version", version)	1✔
22	self._wfsnamespace = namespace	1✔
23	self._query = None	1✔
24
25	def _create_query(self, typename):	1✔
26	self._query = etree.SubElement(self._root, util.nspath('Query', self._wfsnamespace))	1✔
27
28	def set_featureversion(self, version):	1✔
29	self._query.set("featureVersion", version)	×
30
31	def set_propertyname(self, propertyname):	1✔
32	"""Set which feature properties will be returned.
33
34	If not set, will return all properties."""
35	for pn in propertyname:	1✔
36	etree.SubElement(self._query, "PropertyName").text = pn	1✔
37
38	def set_startindex(self, startindex):	1✔
39	"""Set the starting index value for the request"""
40	self._root.set("startIndex", str(startindex))	1✔
41
42	def to_string(self):	1✔
43	"""Returns the xml request in string format.
44
45	Required in order to use the request with getfeature()"""
46	return etree.tostring(self._root)	×
47
48
49	class PostRequest_1_1_0(PostRequest):	1✔
50	"""XML Post request payload builder for WFS version 1.1.0"""
51
52	def __init__(self):	1✔
53	super().__init__(version='1.1.0', namespace=WFS_NAMESPACE)	1✔
54
55	def create_query(self, typename):	1✔
56	"""Creates the query tag with the corresponding typenames.
57	Required element for each request."""
58	super()._create_query(typename)	1✔
59	self._query.set("typeName", typename)	1✔
60
61	def set_bbox(self, bbox):	1✔
62	"""Set a bbox filter.
63
64	Cannot be used with set_featureid() or set_filter().
65	"""
66	filter_tree = etree.SubElement(self._query, util.nspath('Filter', OGC_NAMESPACE))	1✔
67	bbox_tree = etree.SubElement(filter_tree, util.nspath('BBOX', OGC_NAMESPACE))	1✔
68	coords = etree.SubElement(bbox_tree, util.nspath('Envelope', GML_NAMESPACE))	1✔
69	if len(bbox) > 4:	1✔
70	coords.set('srsName', bbox[4])	×
71	lower = etree.SubElement(coords, util.nspath('lowerCorner', GML_NAMESPACE))	1✔
72	lower.text = '{} {}'.format(bbox[0], bbox[1])	1✔
73
74	upper = etree.SubElement(coords, util.nspath('upperCorner', GML_NAMESPACE))	1✔
75	upper.text = '{} {}'.format(bbox[2], bbox[3])	1✔
76
77	def set_featureid(self, featureid):	1✔
78	"""Set filter by feature id.
79
80	Cannot be used with set_bbox() or set_filter().
81	"""
82	feature_tree = etree.SubElement(self._query, util.nspath('Filter', OGC_NAMESPACE))	1✔
83
84	for ft in featureid:	1✔
85	prop_id = etree.Element(util.nspath('GmlObjectId', OGC_NAMESPACE))	1✔
86	prop_id.set(util.nspath('id', GML_NAMESPACE), ft)	1✔
87	feature_tree.append(prop_id)	1✔
88
89	def set_filter(self, filter):	1✔
90	"""Set filter from existing filter.
91
92	Will integrate the filter tag of a provided xml filter to the query being built.
93
94	Cannot be used with set_bbox() or set_featureid().
95	"""
96	f = etree.fromstring(filter)	1✔
97	sub_elem = f.find(util.nspath("Filter", OGC_NAMESPACE))	1✔
98	self._query.append(sub_elem)	1✔
99
100	def set_maxfeatures(self, maxfeatures):	1✔
101	"""Set the maximum number of features to be returned."""
102	self._root.set("maxFeatures", str(maxfeatures))	1✔
103
104	def set_outputformat(self, outputFormat):	1✔
105	"""Set the output format.
106
107	Verify the available formats with a GetCapabilites request."""
108	self._root.set("outputFormat", outputFormat)	1✔
109
110	def set_sortby(self, sortby):	1✔
111	"""Set the properties by which the response will be sorted."""
112	sort_tree = etree.SubElement(self._query, util.nspath("SortBy", OGC_NAMESPACE))	1✔
113	for s in sortby:	1✔
114	prop_elem = etree.SubElement(sort_tree, util.nspath("SortProperty", OGC_NAMESPACE))	1✔
115	prop_name = etree.SubElement(prop_elem, util.nspath('PropertyName', OGC_NAMESPACE))	1✔
116	prop_name.text = s	1✔
117
118
119	class PostRequest_2_0_0(PostRequest):	1✔
120	"""XML Post request payload builder for WFS version 2.0.0."""
121
122	def __init__(self):	1✔
123	super().__init__(version='2.0.0', namespace=WFS20_NAMESPACE)	1✔
124
125	def create_query(self, typename):	1✔
126	"""Creates the query tag with the corresponding typenames.
127	Required element for each request ecept for stored queries."""
128	super()._create_query(typename)	1✔
129	self._query.set("typenames", typename)	1✔
130
131	def create_storedquery(self, stored_id, parameters):	1✔
132	"""Create the storedQuery tag and configure it's sub elements and attributes."""
133	storedquery = etree.SubElement(self._root, util.nspath('StoredQuery', self._wfsnamespace))	1✔
134	storedquery.set("id", str(stored_id))	1✔
135	for param in parameters:	1✔
136	p = etree.SubElement(storedquery, util.nspath('Parameter', self._wfsnamespace))	1✔
137	p.set("name", param)	1✔
138	p.text = parameters[param]	1✔
139
140	def set_bbox(self, bbox):	1✔
141	"""Set a bbox filter.
142
143	Cannot be used with set_featureid() or set_filter().
144	"""
145	filter_tree = etree.SubElement(self._query, util.nspath('Filter', FES_NAMESPACE))	1✔
146	bbox_tree = etree.SubElement(filter_tree, util.nspath('BBOX', FES_NAMESPACE))	1✔
147	coords = etree.SubElement(bbox_tree, util.nspath('Envelope', GML32_NAMESPACE))	1✔
148	if len(bbox) > 4:	1✔
149	coords.set('srsName', bbox[4])	×
150
151	lower = etree.SubElement(coords, util.nspath('lowerCorner', GML32_NAMESPACE))	1✔
152	lower.text = '{} {}'.format(bbox[0], bbox[1])	1✔
153
154	upper = etree.SubElement(coords, util.nspath('upperCorner', GML32_NAMESPACE))	1✔
155	upper.text = '{} {}'.format(bbox[2], bbox[3])	1✔
156
157	def set_featureid(self, featureid):	1✔
158	"""Set filter by feature id.
159
160	Cannot be used with set_bbox() or set_filter().
161	"""
162	feature_tree = etree.SubElement(self._query, util.nspath('Filter', FES_NAMESPACE))	1✔
163	for ft in featureid:	1✔
164	prop_id = etree.Element(util.nspath('ResourceId', FES_NAMESPACE))	1✔
165	prop_id.set('rid', ft)	1✔
166	feature_tree.append(prop_id)	1✔
167
168	def set_filter(self, filter):	1✔
169	"""Set filter from existing filter.
170
171	Will integrate the filter tag of a provided xml filter to the current one
172	being built.
173
174	Cannot be used with set_bbox() or set_featureid().
175	"""
176	if isinstance(filter, str):	1✔
177	f = etree.fromstring(filter)	1✔
178	# If this raises an error, it means the filter string does not contain an actual Filter node.
179	sub_elem = f.xpath("//FES:Filter", namespaces={"FES": FES_NAMESPACE})[0]	1✔
180
181	else:
182	sub_elem = filter	×
183
184	self._query.append(sub_elem)	1✔
185
186	def set_maxfeatures(self, maxfeatures):	1✔
187	"""Set the maximum number of features to be returned."""
188	self._root.set("count", str(maxfeatures))	1✔
189
190	def set_outputformat(self, outputFormat):	1✔
191	"""Set the output format.
192
193	Verify the available formats with a GetCapabilites request.
194	"""
195	self._root.set("outputformat", outputFormat)	1✔
196
197	def set_sortby(self, sortby):	1✔
198	"""Set the properties by which the response will be sorted."""
199	sort_tree = etree.SubElement(self._query, util.nspath("SortBy", FES_NAMESPACE))	1✔
200	for s in sortby:	1✔
201	prop_elem = etree.SubElement(sort_tree, util.nspath("SortProperty", FES_NAMESPACE))	1✔
202	value = etree.SubElement(prop_elem, util.nspath('ValueReference', FES_NAMESPACE))	1✔
203	value.text = s	1✔

geopython / OWSLib / 3903887078

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous