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

82.5

/owslib/feature/schema.py

# =============================================================================
# OWSLib. Copyright (C) 2015 Jachym Cepicky
#
# Contact email: jachym.cepicky@gmail.com
#
# =============================================================================
"""
Set of functions, which are suitable for DescribeFeatureType parsing and
generating layer schema description compatible with `fiona`
"""

import sys
from urllib.parse import urlencode, parse_qsl
from owslib.etree import etree
from owslib.namespaces import Namespaces
from owslib.util import which_etree, findall, Authentication, openURL

MYNS = Namespaces()
XS_NAMESPACE = MYNS.get_namespace("xs")
GML_NAMESPACES = (
    MYNS.get_namespace("gml"),
    MYNS.get_namespace("gml311"),
    MYNS.get_namespace("gml32"),
)


def get_schema(
    url, typename, version="1.0.0", timeout=30, headers=None, username=None, password=None, auth=None
):
    """Parses DescribeFeatureType response and creates schema compatible
    with :class:`fiona`

    :param str url: url of the service
    :param str version: version of the service
    :param str typename: name of the layer
    :param int timeout: request timeout
    :param str username: service authentication username
    :param str password: service authentication password
    :param Authentication auth: instance of owslib.util.Authentication
    """
    if auth:
        if username:
            auth.username = username
        if password:
            auth.password = password
    else:
        auth = Authentication(username, password)
    url = _get_describefeaturetype_url(url, version, typename)
    root = _get_remote_describefeaturetype(url, timeout=timeout,
                                           headers=headers, auth=auth)

    if ":" in typename:
        typename = typename.split(":")[1]
    type_element = root.find("./{%s}element" % XS_NAMESPACE)
    if type_element is None:
        return None
    complex_type = type_element.attrib["type"].split(":")[1]
    elements = _get_elements(complex_type, root)
    nsmap = None
    if hasattr(root, "nsmap"):
        nsmap = root.nsmap
    return _construct_schema(elements, nsmap)


def _get_elements(complex_type, root):
    """Get attribute elements
    """

    found_elements = []
    element = findall(
        root,
        "{%s}complexType" % XS_NAMESPACE,
        attribute_name="name",
        attribute_value=complex_type,
    )[0]
    found_elements = findall(element, "{%s}element" % XS_NAMESPACE)

    return found_elements


def _construct_schema(elements, nsmap):
    """Consruct fiona schema based on given elements

    :param list Element: list of elements
    :param dict nsmap: namespace map

    :return dict: schema
    """
    if elements is None:
        return None
    schema = {"properties": {}, "required": [], "geometry": None}

    schema_key = None
    gml_key = None

    # if nsmap is defined, use it
    if nsmap:
        for key in nsmap:
            if nsmap[key] == XS_NAMESPACE:
                schema_key = key
            if nsmap[key] in GML_NAMESPACES:
                gml_key = key
    # if no nsmap is defined, we have to guess
    else:
        gml_key = "gml"
        schema_key = "xsd"

    mappings = {
        "PointPropertyType": "Point",
        "PolygonPropertyType": "Polygon",
        "LineStringPropertyType": "LineString",
        "MultiPointPropertyType": "MultiPoint",
        "MultiLineStringPropertyType": "MultiLineString",
        "MultiPolygonPropertyType": "MultiPolygon",
        "MultiGeometryPropertyType": "MultiGeometry",
        "GeometryPropertyType": "GeometryCollection",
        "SurfacePropertyType": "3D Polygon",
        "MultiSurfacePropertyType": "3D MultiPolygon",
    }

    for element in elements:
        data_type = element.attrib["type"].replace(gml_key + ":", "")
        name = element.attrib["name"]
        non_nillable = element.attrib.get("nillable", "false") == "false"

        if data_type in mappings:
            schema["geometry"] = mappings[data_type]
            schema["geometry_column"] = name
        else:
            if schema_key is not None:
                schema["properties"][name] = data_type.replace(schema_key + ":", "")

        if non_nillable:
            schema["required"].append(name)

    if schema["properties"] or schema["geometry"]:
        return schema
    else:
        return None


def _get_describefeaturetype_url(url, version, typename):
    """Get url for describefeaturetype request

    :return str: url
    """

    query_string = []
    if url.find("?") != -1:
        query_string = parse_qsl(url.split("?")[1])

    params = [x[0] for x in query_string]

    if "service" not in params:
        query_string.append(("service", "WFS"))
    if "request" not in params:
        query_string.append(("request", "DescribeFeatureType"))
    if "version" not in params:
        query_string.append(("version", version))

    query_string.append(("typeName", typename))

    urlqs = urlencode(tuple(query_string))
    return url.split("?")[0] + "?" + urlqs


def _get_remote_describefeaturetype(url, timeout, headers, auth):
    """Gets the DescribeFeatureType response from the remote server.

    :param str url: url of the service
    :param int timeout: request timeout
    :param Authentication auth: instance of owslib.util.Authentication

    :return etree.Element with the root of the DescribeFeatureType response
    """
    res = openURL(url, timeout=timeout, headers=headers, auth=auth)
    return etree.fromstring(res.read())

1	# =============================================================================
2	# OWSLib. Copyright (C) 2015 Jachym Cepicky
3	#
4	# Contact email: jachym.cepicky@gmail.com
5	#
6	# =============================================================================
7	"""	1✔
8	Set of functions, which are suitable for DescribeFeatureType parsing and
9	generating layer schema description compatible with `fiona`
10	"""
11
12	import sys	1✔
13	from urllib.parse import urlencode, parse_qsl	1✔
14	from owslib.etree import etree	1✔
15	from owslib.namespaces import Namespaces	1✔
16	from owslib.util import which_etree, findall, Authentication, openURL	1✔
17
18	MYNS = Namespaces()	1✔
19	XS_NAMESPACE = MYNS.get_namespace("xs")	1✔
20	GML_NAMESPACES = (	1✔
21	MYNS.get_namespace("gml"),
22	MYNS.get_namespace("gml311"),
23	MYNS.get_namespace("gml32"),
24	)
25
26
27	def get_schema(	1✔
28	url, typename, version="1.0.0", timeout=30, headers=None, username=None, password=None, auth=None
29	):
30	"""Parses DescribeFeatureType response and creates schema compatible
31	with :class:`fiona`
32
33	:param str url: url of the service
34	:param str version: version of the service
35	:param str typename: name of the layer
36	:param int timeout: request timeout
37	:param str username: service authentication username
38	:param str password: service authentication password
39	:param Authentication auth: instance of owslib.util.Authentication
40	"""
41	if auth:	1✔
42	if username:	1✔
43	auth.username = username	×
44	if password:	1✔
45	auth.password = password	×
46	else:
47	auth = Authentication(username, password)	×
48	url = _get_describefeaturetype_url(url, version, typename)	1✔
49	root = _get_remote_describefeaturetype(url, timeout=timeout,	1✔
50	headers=headers, auth=auth)
51
52	if ":" in typename:	1✔
53	typename = typename.split(":")[1]	1✔
54	type_element = root.find("./{%s}element" % XS_NAMESPACE)	1✔
55	if type_element is None:	1✔
56	return None	×
57	complex_type = type_element.attrib["type"].split(":")[1]	1✔
58	elements = _get_elements(complex_type, root)	1✔
59	nsmap = None	1✔
60	if hasattr(root, "nsmap"):	1✔
61	nsmap = root.nsmap	×
62	return _construct_schema(elements, nsmap)	1✔
63
64
65	def _get_elements(complex_type, root):	1✔
66	"""Get attribute elements
67	"""
68
69	found_elements = []	1✔
70	element = findall(	1✔
71	root,
72	"{%s}complexType" % XS_NAMESPACE,
73	attribute_name="name",
74	attribute_value=complex_type,
75	)[0]
76	found_elements = findall(element, "{%s}element" % XS_NAMESPACE)	1✔
77
78	return found_elements	1✔
79
80
81	def _construct_schema(elements, nsmap):	1✔
82	"""Consruct fiona schema based on given elements
83
84	:param list Element: list of elements
85	:param dict nsmap: namespace map
86
87	:return dict: schema
88	"""
89	if elements is None:	1✔
90	return None	×
91	schema = {"properties": {}, "required": [], "geometry": None}	1✔
92
93	schema_key = None	1✔
94	gml_key = None	1✔
95
96	# if nsmap is defined, use it
97	if nsmap:	1✔
98	for key in nsmap:	×
99	if nsmap[key] == XS_NAMESPACE:	×
100	schema_key = key	×
101	if nsmap[key] in GML_NAMESPACES:	×
102	gml_key = key	×
103	# if no nsmap is defined, we have to guess
104	else:
105	gml_key = "gml"	1✔
106	schema_key = "xsd"	1✔
107
108	mappings = {	1✔
109	"PointPropertyType": "Point",
110	"PolygonPropertyType": "Polygon",
111	"LineStringPropertyType": "LineString",
112	"MultiPointPropertyType": "MultiPoint",
113	"MultiLineStringPropertyType": "MultiLineString",
114	"MultiPolygonPropertyType": "MultiPolygon",
115	"MultiGeometryPropertyType": "MultiGeometry",
116	"GeometryPropertyType": "GeometryCollection",
117	"SurfacePropertyType": "3D Polygon",
118	"MultiSurfacePropertyType": "3D MultiPolygon",
119	}
120
121	for element in elements:	1✔
122	data_type = element.attrib["type"].replace(gml_key + ":", "")	1✔
123	name = element.attrib["name"]	1✔
124	non_nillable = element.attrib.get("nillable", "false") == "false"	1✔
125
126	if data_type in mappings:	1✔
127	schema["geometry"] = mappings[data_type]	1✔
128	schema["geometry_column"] = name	1✔
129	else:
130	if schema_key is not None:	1✔
131	schema["properties"][name] = data_type.replace(schema_key + ":", "")	1✔
132
133	if non_nillable:	1✔
134	schema["required"].append(name)	×
135
136	if schema["properties"] or schema["geometry"]:	1✔
137	return schema	1✔
138	else:
139	return None	×
140
141
142	def _get_describefeaturetype_url(url, version, typename):	1✔
143	"""Get url for describefeaturetype request
144
145	:return str: url
146	"""
147
148	query_string = []	1✔
149	if url.find("?") != -1:	1✔
150	query_string = parse_qsl(url.split("?")[1])	×
151
152	params = [x[0] for x in query_string]	1✔
153
154	if "service" not in params:	1✔
155	query_string.append(("service", "WFS"))	1✔
156	if "request" not in params:	1✔
157	query_string.append(("request", "DescribeFeatureType"))	1✔
158	if "version" not in params:	1✔
159	query_string.append(("version", version))	1✔
160
161	query_string.append(("typeName", typename))	1✔
162
163	urlqs = urlencode(tuple(query_string))	1✔
164	return url.split("?")[0] + "?" + urlqs	1✔
165
166
167	def _get_remote_describefeaturetype(url, timeout, headers, auth):	1✔
168	"""Gets the DescribeFeatureType response from the remote server.
169
170	:param str url: url of the service
171	:param int timeout: request timeout
172	:param Authentication auth: instance of owslib.util.Authentication
173
174	:return etree.Element with the root of the DescribeFeatureType response
175	"""
176	res = openURL(url, timeout=timeout, headers=headers, auth=auth)	1✔
177	return etree.fromstring(res.read())	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