3663701074

Build Type

github

Committed by GitHub

Commit Message

Merge 6cd54d613 into 13b1443f7

Pull Request Pull Request #851: Adding Python 3.10 in CI

Run Details

7461 of 12701 relevant lines covered (58.74%)

0.59 hits per line

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

36.84

/owslib/feature/__init__.py

# =============================================================================
# OWSLib. Copyright (C) 2012 Jachym Cepicky
#
# Contact email: jachym.cepicky@gmail.com
#
# =============================================================================

from urllib.parse import urlencode
from owslib.crs import Crs
from owslib.util import log, Authentication
from owslib.feature.schema import get_schema
from owslib.feature.postrequest import PostRequest_1_1_0, PostRequest_2_0_0


class WebFeatureService_(object):
    """Base class for WebFeatureService implementations"""

    def __init__(self, auth=None):
        self.auth = auth or Authentication()

    def getBBOXKVP(self, bbox, typename):
        """Formate bounding box for KVP request type (HTTP GET)

        @param bbox: (minx,miny,maxx,maxy[,srs])
        @type bbox: List
        @param typename:  feature name
        @type typename: String
        @returns: String properly formated according to version and
            coordinate reference system
        """
        srs = None

        # srs of the bbox is specified in the bbox as fifth paramter
        if len(bbox) == 5:
            srs = Crs(bbox[4])
        # take default srs
        else:
            srs = self.contents[typename[0]].crsOptions[0]

        # 1.1.0 and 2.0.0 have same encoding
        if self.version in ["1.1.0", "2.0.0"]:

            # format bbox parameter
            if srs.encoding == "urn":
                if srs.axisorder == "yx":
                    return "%s,%s,%s,%s,%s" % (
                        bbox[1],
                        bbox[0],
                        bbox[3],
                        bbox[2],
                        srs.getcodeurn(),
                    )
                else:
                    return "%s,%s,%s,%s,%s" % (
                        bbox[0],
                        bbox[1],
                        bbox[2],
                        bbox[3],
                        srs.getcodeurn(),
                    )
            else:
                return "%s,%s,%s,%s,%s" % (
                    bbox[0],
                    bbox[1],
                    bbox[2],
                    bbox[3],
                    srs.getcode(),
                )
        # 1.0.0
        else:
            return "%s,%s,%s,%s,%s" % (
                bbox[0],
                bbox[1],
                bbox[2],
                bbox[3],
                srs.getcode(),
            )

    def getBBOXPost(self, bbox, typename):
        """Format bounding box for Post requests

        @param bbox: (minx,miny,maxx,maxy[,srs])
        @type bbox: List
        @param typename:  feature name
        @type typename: String
        @returns: String properly formated according to version and
            coordinate reference system
        """
        srs = None

        # srs of the bbox is specified in the bbox as fifth paramter
        if len(bbox) == 5:
            srs = Crs(bbox[4])
        # take default srs
        else:
            srs = self.contents[typename[0]].crsOptions[0]

        formatted_bbox = [bbox[0], bbox[1], bbox[2], bbox[3]]
        if self.version in ["1.1.0", "2.0.0"]:
            if srs.axisorder == "yx" and srs.encoding == "urn":
                formatted_bbox = [bbox[1], bbox[0], bbox[3], bbox[2]]

            if self.version == "1.1.0":
                formatted_bbox.append(srs.getcodeurn())
                return formatted_bbox
            if self.version == "2.0.0":
                formatted_bbox.append(srs.getcodeuri1())
                return formatted_bbox
        else:
            formatted_bbox.append(srs.getcode())
            return formatted_bbox

    def create_post_request(self):
        """Creates an xml POST request according to WFS version."""

        if self.version in ['1.1.0']:
            return PostRequest_1_1_0()

        if self.version in ['2.0', '2.0.0']:
            return PostRequest_2_0_0()

    def getSRS(self, srsname, typename):
        """Returns None or Crs object for given name

        @param typename:  feature name
        @type typename: String
        """
        if not isinstance(srsname, Crs):
            srs = Crs(srsname)
        else:
            srs = srsname

        try:
            index = self.contents[typename].crsOptions.index(srs)
            # Return the Crs string that was pulled directly from the
            # GetCaps document (the 'id' attribute in the Crs object).
            return self.contents[typename].crsOptions[index]
        except ValueError:
            options = ", ".join([crs.id for crs in self.contents[typename].crsOptions])
            log.warning(
                "Requested srsName %r is not declared as being "
                "allowed for requested typename %r. "
                "Options are: %r.",
                srs.getcode(),
                typename,
                options,
            )
            return None

    def getGETGetFeatureRequest(
        self,
        typename=None,
        filter=None,
        bbox=None,
        featureid=None,
        featureversion=None,
        propertyname=None,
        maxfeatures=None,
        storedQueryID=None,
        storedQueryParams=None,
        outputFormat=None,
        method="Get",
        startindex=None,
        sortby=None,
    ):
        """Formulate proper GetFeature request using KVP encoding
        ----------
        typename : list
            List of typenames (string)
        filter : string
            XML-encoded OGC filter expression.
        bbox : tuple
            (left, bottom, right, top) in the feature type's coordinates == (minx, miny, maxx, maxy)
        featureid : list
            List of unique feature ids (string)
        featureversion : string
            Default is most recent feature version.
        propertyname : list
            List of feature property names. '*' matches all.
        maxfeatures : int
            Maximum number of features to be returned.
        method : string
            Qualified name of the HTTP DCP method to use.
        outputFormat: string (optional)
            Requested response format of the request.
        startindex: int (optional)
            Start position to return feature set (paging in combination with maxfeatures)
        sortby: list (optional)
            List of property names whose values should be used to order
            (upon presentation) the set of feature instances that
            satify the query.

        There are 3 different modes of use

        1) typename and bbox (simple spatial query)
        2) typename and filter (==query) (more expressive)
        3) featureid (direct access to known features)
        """
        storedQueryParams = storedQueryParams or {}

        base_url = next(
            (
                m.get("url")
                for m in self.getOperationByName("GetFeature").methods
                if m.get("type").lower() == method.lower()
            )
        )
        base_url = base_url if base_url.endswith("?") else base_url + "?"

        request = {"service": "WFS", "version": self.version, "request": "GetFeature"}

        # check featureid
        if featureid:
            request["featureid"] = ",".join(featureid)
        elif bbox:
            request["bbox"] = self.getBBOXKVP(bbox, typename)
        elif filter:
            request["query"] = str(filter)
        if typename:
            typename = (
                [typename] if isinstance(typename, str) else typename
            )  # noqa: E721
            if int(self.version.split(".")[0]) >= 2:
                request["typenames"] = ",".join(typename)
            else:
                request["typename"] = ",".join(typename)
        if propertyname:
            request["propertyname"] = ",".join(propertyname)
        if sortby:
            request["sortby"] = ",".join(sortby)
        if featureversion:
            request["featureversion"] = str(featureversion)
        if maxfeatures:
            if int(self.version.split(".")[0]) >= 2:
                request["count"] = str(maxfeatures)
            else:
                request["maxfeatures"] = str(maxfeatures)
        if startindex:
            request["startindex"] = str(startindex)
        if storedQueryID:
            request["storedQuery_id"] = str(storedQueryID)
            for param in storedQueryParams:
                request[param] = storedQueryParams[param]
        if outputFormat is not None:
            request["outputFormat"] = outputFormat

        data = urlencode(request, doseq=True)

        return base_url + data

    def getPOSTGetFeatureRequest(
        self,
        typename=None,
        filter=None,
        bbox=None,
        featureid=None,
        featureversion=None,
        propertyname=None,
        maxfeatures=None,
        storedQueryID=None,
        storedQueryParams=None,
        outputFormat=None,
        method="Post",
        startindex=None,
        sortby=None,
    ):
        """Formulate proper GetFeature request using KVP encoding
        ----------
        typename : list
            List of typenames (string)
        filter : string
            XML-encoded OGC filter expression.
        bbox : tuple
            (left, bottom, right, top) in the feature type's coordinates == (minx, miny, maxx, maxy)
        featureid : list
            List of unique feature ids (string)
        featureversion : string
            Default is most recent feature version.
        propertyname : list
            List of feature property names. Leave blank (None) to get all properties.
        maxfeatures : int
            Maximum number of features to be returned.
        method : string
            Qualified name of the HTTP DCP method to use.
        outputFormat: string (optional)
            Requested response format of the request.
        startindex: int (optional)
            Start position to return feature set (paging in combination with maxfeatures)
        storedQueryID : string
            A name identifying a prepared set available in WFS-service.
            WFS version 2.0.0 and above only.
        storedQueryParams : dict
            Variable amount of extra information sent to server related to
            storedQueryID to further define the requested data.
            WFS version 2.0.0 and above only.
            {'parameter_name': parameter_value}
        sortby: list (optional)
            List of property names whose values should be used to order
            (upon presentation) the set of feature instances that
            satify the query.

        There are 5 different modes of use

        1) typename and bbox (simple spatial query)
        2) typename and filter (==query) (more expressive)
        3) featureid (direct access to known features)
        4) storedQueryID and optional storedQueryParams
        5) filter only via Post method
        """

        try:
            base_url = next(
                (
                    m.get("url")
                    for m in self.getOperationByName("GetFeature").methods
                    if m.get("type").lower() == method.lower()
                )
            )
        except StopIteration:
            base_url = self.url

        if not typename and filter:
            return base_url, filter

        request = self.create_post_request()

        if storedQueryID:
            if self.version in ["1.0.0", "1.1.0"]:
                log.warning("Stored queries are only supported in version 2.0.0 and above.")
                return None

            storedQueryParams = storedQueryParams or {}
            request.create_storedquery(storedQueryID, storedQueryParams)
            data = request.to_string()
            return base_url, data

        typename = (
            [typename] if isinstance(typename, str) else typename
        )  # noqa: E721
        typenames = ",".join(typename)

        request.create_query(typenames)

        if featureid:
            featureid = (
                [featureid] if isinstance(featureid, str) else featureid
            )
            request.set_featureid(featureid)
        elif bbox:
            request.set_bbox(self.getBBOXPost(bbox, typename))
        elif filter:
            request.set_filter(filter)

        if featureversion:
            request.set_featureversion(str(featureversion))
        if maxfeatures:
            request.set_maxfeatures(maxfeatures)
        if outputFormat:
            request.set_outputformat(outputFormat)
        if propertyname:
            propertyname = (
                [propertyname] if isinstance(propertyname, str) else propertyname
            )
            request.set_propertyname(propertyname)
        if sortby:
            sortby = (
                [sortby] if isinstance(sortby, str) else sortby
            )
            request.set_sortby(sortby)
        if startindex:
            request.set_startindex(startindex)

        data = request.to_string()
        return base_url, data

    def get_schema(self, typename):
        """
        Get layer schema compatible with :class:`fiona` schema object
        """
        return get_schema(self.url, typename, self.version, auth=self.auth)

1	# =============================================================================
2	# OWSLib. Copyright (C) 2012 Jachym Cepicky
3	#
4	# Contact email: jachym.cepicky@gmail.com
5	#
6	# =============================================================================
7
8	from urllib.parse import urlencode	1✔
9	from owslib.crs import Crs	1✔
10	from owslib.util import log, Authentication	1✔
11	from owslib.feature.schema import get_schema	1✔
12	from owslib.feature.postrequest import PostRequest_1_1_0, PostRequest_2_0_0	1✔
13
14
15	class WebFeatureService_(object):	1✔
16	"""Base class for WebFeatureService implementations"""
17
18	def __init__(self, auth=None):	1✔
19	self.auth = auth or Authentication()	1✔
20
21	def getBBOXKVP(self, bbox, typename):	1✔
22	"""Formate bounding box for KVP request type (HTTP GET)
23
24	@param bbox: (minx,miny,maxx,maxy[,srs])
25	@type bbox: List
26	@param typename: feature name
27	@type typename: String
28	@returns: String properly formated according to version and
29	coordinate reference system
30	"""
31	srs = None	1✔
32
33	# srs of the bbox is specified in the bbox as fifth paramter
34	if len(bbox) == 5:	1✔
35	srs = Crs(bbox[4])	1✔
36	# take default srs
37	else:
38	srs = self.contents[typename[0]].crsOptions[0]	1✔
39
40	# 1.1.0 and 2.0.0 have same encoding
41	if self.version in ["1.1.0", "2.0.0"]:	1✔
42
43	# format bbox parameter
44	if srs.encoding == "urn":	1✔
45	if srs.axisorder == "yx":	1✔
46	return "%s,%s,%s,%s,%s" % (	1✔
47	bbox[1],
48	bbox[0],
49	bbox[3],
50	bbox[2],
51	srs.getcodeurn(),
52	)
53	else:
54	return "%s,%s,%s,%s,%s" % (	×
55	bbox[0],
56	bbox[1],
57	bbox[2],
58	bbox[3],
59	srs.getcodeurn(),
60	)
61	else:
62	return "%s,%s,%s,%s,%s" % (	×
63	bbox[0],
64	bbox[1],
65	bbox[2],
66	bbox[3],
67	srs.getcode(),
68	)
69	# 1.0.0
70	else:
71	return "%s,%s,%s,%s,%s" % (	×
72	bbox[0],
73	bbox[1],
74	bbox[2],
75	bbox[3],
76	srs.getcode(),
77	)
78
79	def getBBOXPost(self, bbox, typename):	1✔
80	"""Format bounding box for Post requests
81
82	@param bbox: (minx,miny,maxx,maxy[,srs])
83	@type bbox: List
84	@param typename: feature name
85	@type typename: String
86	@returns: String properly formated according to version and
87	coordinate reference system
88	"""
89	srs = None	×
90
91	# srs of the bbox is specified in the bbox as fifth paramter
92	if len(bbox) == 5:	×
93	srs = Crs(bbox[4])	×
94	# take default srs
95	else:
96	srs = self.contents[typename[0]].crsOptions[0]	×
97
98	formatted_bbox = [bbox[0], bbox[1], bbox[2], bbox[3]]	×
99	if self.version in ["1.1.0", "2.0.0"]:	×
100	if srs.axisorder == "yx" and srs.encoding == "urn":	×
101	formatted_bbox = [bbox[1], bbox[0], bbox[3], bbox[2]]	×
102
103	if self.version == "1.1.0":	×
104	formatted_bbox.append(srs.getcodeurn())	×
105	return formatted_bbox	×
106	if self.version == "2.0.0":	×
107	formatted_bbox.append(srs.getcodeuri1())	×
108	return formatted_bbox	×
109	else:
110	formatted_bbox.append(srs.getcode())	×
111	return formatted_bbox	×
112
113	def create_post_request(self):	1✔
114	"""Creates an xml POST request according to WFS version."""
115
116	if self.version in ['1.1.0']:	×
117	return PostRequest_1_1_0()	×
118
119	if self.version in ['2.0', '2.0.0']:	×
120	return PostRequest_2_0_0()	×
121
122	def getSRS(self, srsname, typename):	1✔
123	"""Returns None or Crs object for given name
124
125	@param typename: feature name
126	@type typename: String
127	"""
128	if not isinstance(srsname, Crs):	×
129	srs = Crs(srsname)	×
130	else:
131	srs = srsname	×
132
133	try:	×
134	index = self.contents[typename].crsOptions.index(srs)	×
135	# Return the Crs string that was pulled directly from the
136	# GetCaps document (the 'id' attribute in the Crs object).
137	return self.contents[typename].crsOptions[index]	×
138	except ValueError:	×
139	options = ", ".join([crs.id for crs in self.contents[typename].crsOptions])	×
140	log.warning(	×
141	"Requested srsName %r is not declared as being "
142	"allowed for requested typename %r. "
143	"Options are: %r.",
144	srs.getcode(),
145	typename,
146	options,
147	)
148	return None	×
149
150	def getGETGetFeatureRequest(	1✔
151	self,
152	typename=None,
153	filter=None,
154	bbox=None,
155	featureid=None,
156	featureversion=None,
157	propertyname=None,
158	maxfeatures=None,
159	storedQueryID=None,
160	storedQueryParams=None,
161	outputFormat=None,
162	method="Get",
163	startindex=None,
164	sortby=None,
165	):
166	"""Formulate proper GetFeature request using KVP encoding
167	----------
168	typename : list
169	List of typenames (string)
170	filter : string
171	XML-encoded OGC filter expression.
172	bbox : tuple
173	(left, bottom, right, top) in the feature type's coordinates == (minx, miny, maxx, maxy)
174	featureid : list
175	List of unique feature ids (string)
176	featureversion : string
177	Default is most recent feature version.
178	propertyname : list
179	List of feature property names. '*' matches all.
180	maxfeatures : int
181	Maximum number of features to be returned.
182	method : string
183	Qualified name of the HTTP DCP method to use.
184	outputFormat: string (optional)
185	Requested response format of the request.
186	startindex: int (optional)
187	Start position to return feature set (paging in combination with maxfeatures)
188	sortby: list (optional)
189	List of property names whose values should be used to order
190	(upon presentation) the set of feature instances that
191	satify the query.
192
193	There are 3 different modes of use
194
195	1) typename and bbox (simple spatial query)
196	2) typename and filter (==query) (more expressive)
197	3) featureid (direct access to known features)
198	"""
199	storedQueryParams = storedQueryParams or {}	1✔
200
201	base_url = next(	1✔
202	(
203	m.get("url")
204	for m in self.getOperationByName("GetFeature").methods
205	if m.get("type").lower() == method.lower()
206	)
207	)
208	base_url = base_url if base_url.endswith("?") else base_url + "?"	1✔
209
210	request = {"service": "WFS", "version": self.version, "request": "GetFeature"}	1✔
211
212	# check featureid
213	if featureid:	1✔
214	request["featureid"] = ",".join(featureid)	×
215	elif bbox:	1✔
216	request["bbox"] = self.getBBOXKVP(bbox, typename)	1✔
217	elif filter:	1✔
218	request["query"] = str(filter)	1✔
219	if typename:	1✔
220	typename = (	1✔
221	[typename] if isinstance(typename, str) else typename
222	) # noqa: E721
223	if int(self.version.split(".")[0]) >= 2:	1✔
224	request["typenames"] = ",".join(typename)	1✔
225	else:
226	request["typename"] = ",".join(typename)	1✔
227	if propertyname:	1✔
228	request["propertyname"] = ",".join(propertyname)	×
229	if sortby:	1✔
230	request["sortby"] = ",".join(sortby)	×
231	if featureversion:	1✔
232	request["featureversion"] = str(featureversion)	×
233	if maxfeatures:	1✔
234	if int(self.version.split(".")[0]) >= 2:	1✔
235	request["count"] = str(maxfeatures)	×
236	else:
237	request["maxfeatures"] = str(maxfeatures)	1✔
238	if startindex:	1✔
239	request["startindex"] = str(startindex)	×
240	if storedQueryID:	1✔
241	request["storedQuery_id"] = str(storedQueryID)	×
242	for param in storedQueryParams:	×
243	request[param] = storedQueryParams[param]	×
244	if outputFormat is not None:	1✔
245	request["outputFormat"] = outputFormat	×
246
247	data = urlencode(request, doseq=True)	1✔
248
249	return base_url + data	1✔
250
251	def getPOSTGetFeatureRequest(	1✔
252	self,
253	typename=None,
254	filter=None,
255	bbox=None,
256	featureid=None,
257	featureversion=None,
258	propertyname=None,
259	maxfeatures=None,
260	storedQueryID=None,
261	storedQueryParams=None,
262	outputFormat=None,
263	method="Post",
264	startindex=None,
265	sortby=None,
266	):
267	"""Formulate proper GetFeature request using KVP encoding
268	----------
269	typename : list
270	List of typenames (string)
271	filter : string
272	XML-encoded OGC filter expression.
273	bbox : tuple
274	(left, bottom, right, top) in the feature type's coordinates == (minx, miny, maxx, maxy)
275	featureid : list
276	List of unique feature ids (string)
277	featureversion : string
278	Default is most recent feature version.
279	propertyname : list
280	List of feature property names. Leave blank (None) to get all properties.
281	maxfeatures : int
282	Maximum number of features to be returned.
283	method : string
284	Qualified name of the HTTP DCP method to use.
285	outputFormat: string (optional)
286	Requested response format of the request.
287	startindex: int (optional)
288	Start position to return feature set (paging in combination with maxfeatures)
289	storedQueryID : string
290	A name identifying a prepared set available in WFS-service.
291	WFS version 2.0.0 and above only.
292	storedQueryParams : dict
293	Variable amount of extra information sent to server related to
294	storedQueryID to further define the requested data.
295	WFS version 2.0.0 and above only.
296	{'parameter_name': parameter_value}
297	sortby: list (optional)
298	List of property names whose values should be used to order
299	(upon presentation) the set of feature instances that
300	satify the query.
301
302	There are 5 different modes of use
303
304	1) typename and bbox (simple spatial query)
305	2) typename and filter (==query) (more expressive)
306	3) featureid (direct access to known features)
307	4) storedQueryID and optional storedQueryParams
308	5) filter only via Post method
309	"""
310
311	try:	×
312	base_url = next(	×
313	(
314	m.get("url")
315	for m in self.getOperationByName("GetFeature").methods
316	if m.get("type").lower() == method.lower()
317	)
318	)
319	except StopIteration:	×
320	base_url = self.url	×
321
322	if not typename and filter:	×
323	return base_url, filter	×
324
325	request = self.create_post_request()	×
326
327	if storedQueryID:	×
328	if self.version in ["1.0.0", "1.1.0"]:	×
329	log.warning("Stored queries are only supported in version 2.0.0 and above.")	×
330	return None	×
331
332	storedQueryParams = storedQueryParams or {}	×
333	request.create_storedquery(storedQueryID, storedQueryParams)	×
334	data = request.to_string()	×
335	return base_url, data	×
336
337	typename = (	×
338	[typename] if isinstance(typename, str) else typename
339	) # noqa: E721
340	typenames = ",".join(typename)	×
341
342	request.create_query(typenames)	×
343
344	if featureid:	×
345	featureid = (	×
346	[featureid] if isinstance(featureid, str) else featureid
347	)
348	request.set_featureid(featureid)	×
349	elif bbox:	×
350	request.set_bbox(self.getBBOXPost(bbox, typename))	×
351	elif filter:	×
352	request.set_filter(filter)	×
353
354	if featureversion:	×
355	request.set_featureversion(str(featureversion))	×
356	if maxfeatures:	×
357	request.set_maxfeatures(maxfeatures)	×
358	if outputFormat:	×
359	request.set_outputformat(outputFormat)	×
360	if propertyname:	×
361	propertyname = (	×
362	[propertyname] if isinstance(propertyname, str) else propertyname
363	)
364	request.set_propertyname(propertyname)	×
365	if sortby:	×
366	sortby = (	×
367	[sortby] if isinstance(sortby, str) else sortby
368	)
369	request.set_sortby(sortby)	×
370	if startindex:	×
371	request.set_startindex(startindex)	×
372
373	data = request.to_string()	×
374	return base_url, data	×
375
376	def get_schema(self, typename):	1✔
377	"""
378	Get layer schema compatible with :class:`fiona` schema object
379	"""
380	return get_schema(self.url, typename, self.version, auth=self.auth)	1✔

geopython / OWSLib / 3663701074

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