11628884181

Committed 01 Nov 2024 11:49AM UTC coverage: 58.814% (-1.3%) from 60.156%

Build # 11628884181

Build Type

Pull #548

github

Committed by

web-flow

Commit Message

Merge 4b9b7cf1f into ae98c2039

Pull Request Pull Request #548: Strip out redundant __new__ to enable object pickling

Run Details

8364 of 14221 relevant lines covered (58.81%)

1.18 hits per line

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

54.55

/owslib/ogcapi/features.py

# =============================================================================
# Copyright (c) 2022 Tom Kralidis
#
# Author: Tom Kralidis <tomkralidis@gmail.com>
#
# Contact email: tomkralidis@gmail.com
# =============================================================================

from copy import deepcopy
import logging
from urllib.parse import urlencode

from owslib.ogcapi import Collections
from owslib.util import Authentication

LOGGER = logging.getLogger(__name__)


class Features(Collections):
    """Abstraction for OGC API - Features"""

    def __init__(self, url: str, json_: str = None, timeout: int = 30,
                 headers: dict = None, auth: Authentication = None):
        __doc__ = Collections.__doc__  # noqa
        super().__init__(url, json_, timeout, headers, auth)

    def feature_collections(self) -> list:
        """
        implements /collections filtered on features

        @returns: `list` of filtered collections object
        """

        features_ = []
        collections_ = super().collections()

        for c_ in collections_['collections']:
            if 'itemType' in c_ and c_['itemType'].lower() == 'feature':
                features_.append(c_['id'])

        return features_

    def collection_queryables(self, collection_id: str) -> dict:
        """
        implements /collections/{collectionId}/queryables

        @type collection_id: string
        @param collection_id: id of collection

        @returns: `dict` of feature collection queryables
        """

        path = f'collections/{collection_id}/queryables'
        return self._request(path=path)

    def collection_items(self, collection_id: str, **kwargs: dict) -> dict:
        """
        implements /collection/{collectionId}/items

        @type collection_id: string
        @param collection_id: id of collection
        @type bbox: list
        @param bbox: list of minx,miny,maxx,maxy
        @type datetime_: string
        @param datetime_: time extent or time instant
        @type limit: int
        @param limit: limit number of features
        @type offset: int
        @param offset: start position of results
        @type q: string
        @param q: full text search
        @type filter: string
        @param filter: CQL TEXT expression
        @type cql: dict
        @param cql: CQL JSON payload

        @returns: feature results
        """

        if 'bbox' in kwargs:
            kwargs['bbox'] = ','.join(list(map(str, kwargs['bbox'])))
        if 'datetime_' in kwargs:
            kwargs['datetime'] = kwargs['datetime_']

        if 'cql' in kwargs:
            LOGGER.debug('CQL query detected')
            kwargs2 = deepcopy(kwargs)
            cql = kwargs2.pop('cql')
            path = f'collections/{collection_id}/items?{urlencode(kwargs2)}'
            return self._request(method='POST', path=path, data=cql, kwargs=kwargs2)
        else:
            path = f'collections/{collection_id}/items'
            return self._request(path=path, kwargs=kwargs)

    def collection_item(self, collection_id: str, identifier: str) -> dict:
        """
        implements /collections/{collectionId}/items/{featureId}

        @type collection_id: string
        @param collection_id: id of collection
        @type identifier: string
        @param identifier: feature identifier

        @returns: single feature result
        """

        path = f'collections/{collection_id}/items/{identifier}'
        return self._request(path=path)

    def collection_item_create(self, collection_id: str, data: str) -> bool:
        """
        implements POST /collections/{collectionId}/items

        @type collection_id: string
        @param collection_id: id of collection
        @type data: string
        @param data: raw representation of data

        @returns: single feature result
        """

        path = f'collections/{collection_id}/items'

        if isinstance(data, dict):  # JSON
            LOGGER.debug('Detected JSON payload')
            self.headers['Content-Type'] = 'application/geo+json'
        elif data.startswith('<'):  # XML
            data = data.strip()
            LOGGER.debug('Detected XML payload')
            self.headers['Content-Type'] = 'application/xml'

        _ = self._request(method='POST', path=path, data=data)

        return True

    def collection_item_update(self, collection_id: str, identifier: str,
                               data: str) -> bool:
        """
        implements PUT /collections/{collectionId}/items/{featureId}

        @type collection_id: string
        @param collection_id: id of collection
        @type identifier: string
        @param identifier: feature identifier
        @type data: string
        @param data: raw representation of data

        @returns: ``bool`` of deletion result
        """

        path = f'collections/{collection_id}/items/{identifier}'
        _ = self._request(method='PUT', path=path, data=data)

        return True

    def collection_item_delete(self, collection_id: str, identifier: str) -> bool:
        """
        implements DELETE /collections/{collectionId}/items/{featureId}

        @type collection_id: string
        @param collection_id: id of collection
        @type identifier: string
        @param identifier: feature identifier

        @returns: ``bool`` of deletion result
        """

        path = f'collections/{collection_id}/items/{identifier}'
        _ = self._request(method='DELETE', path=path)

        return True

1	# =============================================================================
2	# Copyright (c) 2022 Tom Kralidis
3	#
4	# Author: Tom Kralidis <tomkralidis@gmail.com>
5	#
6	# Contact email: tomkralidis@gmail.com
7	# =============================================================================
8
9	from copy import deepcopy	2✔
10	import logging	2✔
11	from urllib.parse import urlencode	2✔
12
13	from owslib.ogcapi import Collections	2✔
14	from owslib.util import Authentication	2✔
15
16	LOGGER = logging.getLogger(__name__)	2✔
17
18
19	class Features(Collections):	2✔
20	"""Abstraction for OGC API - Features"""
21
22	def __init__(self, url: str, json_: str = None, timeout: int = 30,	2✔
23	headers: dict = None, auth: Authentication = None):
24	__doc__ = Collections.__doc__ # noqa	2✔
25	super().__init__(url, json_, timeout, headers, auth)	2✔
26
27	def feature_collections(self) -> list:	2✔
28	"""
29	implements /collections filtered on features
30
31	@returns: `list` of filtered collections object
32	"""
33
34	features_ = []	2✔
35	collections_ = super().collections()	2✔
36
37	for c_ in collections_['collections']:	2✔
38	if 'itemType' in c_ and c_['itemType'].lower() == 'feature':	2✔
39	features_.append(c_['id'])	2✔
40
41	return features_	2✔
42
43	def collection_queryables(self, collection_id: str) -> dict:	2✔
44	"""
45	implements /collections/{collectionId}/queryables
46
47	@type collection_id: string
48	@param collection_id: id of collection
49
50	@returns: `dict` of feature collection queryables
51	"""
52
53	path = f'collections/{collection_id}/queryables'	2✔
54	return self._request(path=path)	2✔
55
56	def collection_items(self, collection_id: str, **kwargs: dict) -> dict:	2✔
57	"""
58	implements /collection/{collectionId}/items
59
60	@type collection_id: string
61	@param collection_id: id of collection
62	@type bbox: list
63	@param bbox: list of minx,miny,maxx,maxy
64	@type datetime_: string
65	@param datetime_: time extent or time instant
66	@type limit: int
67	@param limit: limit number of features
68	@type offset: int
69	@param offset: start position of results
70	@type q: string
71	@param q: full text search
72	@type filter: string
73	@param filter: CQL TEXT expression
74	@type cql: dict
75	@param cql: CQL JSON payload
76
77	@returns: feature results
78	"""
79
80	if 'bbox' in kwargs:	2✔
81	kwargs['bbox'] = ','.join(list(map(str, kwargs['bbox'])))	×
82	if 'datetime_' in kwargs:	2✔
83	kwargs['datetime'] = kwargs['datetime_']	×
84
85	if 'cql' in kwargs:	2✔
86	LOGGER.debug('CQL query detected')	×
87	kwargs2 = deepcopy(kwargs)	×
88	cql = kwargs2.pop('cql')	×
89	path = f'collections/{collection_id}/items?{urlencode(kwargs2)}'	×
90	return self._request(method='POST', path=path, data=cql, kwargs=kwargs2)	×
91	else:
92	path = f'collections/{collection_id}/items'	2✔
93	return self._request(path=path, kwargs=kwargs)	2✔
94
95	def collection_item(self, collection_id: str, identifier: str) -> dict:	2✔
96	"""
97	implements /collections/{collectionId}/items/{featureId}
98
99	@type collection_id: string
100	@param collection_id: id of collection
101	@type identifier: string
102	@param identifier: feature identifier
103
104	@returns: single feature result
105	"""
106
107	path = f'collections/{collection_id}/items/{identifier}'	×
108	return self._request(path=path)	×
109
110	def collection_item_create(self, collection_id: str, data: str) -> bool:	2✔
111	"""
112	implements POST /collections/{collectionId}/items
113
114	@type collection_id: string
115	@param collection_id: id of collection
116	@type data: string
117	@param data: raw representation of data
118
119	@returns: single feature result
120	"""
121
122	path = f'collections/{collection_id}/items'	×
123
124	if isinstance(data, dict): # JSON	×
125	LOGGER.debug('Detected JSON payload')	×
126	self.headers['Content-Type'] = 'application/geo+json'	×
127	elif data.startswith('<'): # XML	×
128	data = data.strip()	×
129	LOGGER.debug('Detected XML payload')	×
130	self.headers['Content-Type'] = 'application/xml'	×
131
132	_ = self._request(method='POST', path=path, data=data)	×
133
134	return True	×
135
136	def collection_item_update(self, collection_id: str, identifier: str,	2✔
137	data: str) -> bool:
138	"""
139	implements PUT /collections/{collectionId}/items/{featureId}
140
141	@type collection_id: string
142	@param collection_id: id of collection
143	@type identifier: string
144	@param identifier: feature identifier
145	@type data: string
146	@param data: raw representation of data
147
148	@returns: ``bool`` of deletion result
149	"""
150
151	path = f'collections/{collection_id}/items/{identifier}'	×
152	_ = self._request(method='PUT', path=path, data=data)	×
153
154	return True	×
155
156	def collection_item_delete(self, collection_id: str, identifier: str) -> bool:	2✔
157	"""
158	implements DELETE /collections/{collectionId}/items/{featureId}
159
160	@type collection_id: string
161	@param collection_id: id of collection
162	@type identifier: string
163	@param identifier: feature identifier
164
165	@returns: ``bool`` of deletion result
166	"""
167
168	path = f'collections/{collection_id}/items/{identifier}'	×
169	_ = self._request(method='DELETE', path=path)	×
170
171	return True	×

geopython / OWSLib / 11628884181

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