11282283562

Committed 10 Oct 2024 09:29PM UTC coverage: 60.113% (-0.02%) from 60.128%

Build # 11282283562

Build Type

Pull #949

github

Committed by

web-flow

Commit Message

Merge 1058b0b85 into 852fe6d4b

Pull Request Pull Request #949: Remove dependency on the Pytz library

Run Details

2 of 2 new or added lines in 1 file covered. (100.0%)

82 existing lines in 4 files now uncovered.

8533 of 14195 relevant lines covered (60.11%)

0.6 hits per line

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

45.99

/owslib/feature/__init__.py

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

import logging

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

LOGGER = logging.getLogger(__name__)


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])
            LOGGER.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,
        srsname=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.
        srsname: string
            EPSG code to request the data in
        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()
            )
        )

        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 srsname:
            request["srsname"] = str(srsname)

            # Check if desired SRS is supported by the service for each
            # typename. Warning will be thrown if that SRS is not allowed.
            for name in typename:
                _ = self.getSRS(srsname, name)
        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

        return build_get_url(base_url, request)

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

geopython / OWSLib / 11282283562

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