Merge pull request #20 from docentYT/development

v2.0.0

API.txt

@@ -59,6 +59,7 @@ NOTES
 ✅ POST /api/0.6/notes/#id/comment - Create new comment to note
 ✅❓ POST /api/0.6/notes/#id/close - Close note
 ✅❓ POST /api/0.6/notes/#id/reopen - Reopen note
+✅❓ MO: DELETE /api/0.6/notes/#id/ - Hide note.
 ✅❓ GET /api/0.6/notes/search - Search for notes
 X GET /api/0.6/notes/feed?bbox=Left,Bottom,Right,Top - Get RSS feed for notes in bbox
 

CHANGELOG.md

@@ -5,6 +5,23 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [2.0.0]
+### Added
+- Missing status code handling in `notes.get()`.
+- Support for hide note endpoint.
+### Fixed
+- Problems with parsing data chunks received by api.
+- Small grammar corrections in the documentation.
+### Changed
+- Working method of parser used in `notes` endpoint.
+- Working method of parsers in `misc` endpoint.
+- Working method of parsers in `user` endpoint.
+- Working method of parsers in `changeset` endpoint.
+- Working method of parsers in `elements` endpoint.
+### Removed
+- `EmptyResult` api exception, which was used in endpoints `notes`, `user` and `changeset`. From now on when the results are empty an empty list will be returned.
+- Unused imports.
+
 ## [1.1.1]
 ### Fixed
 - Corrected character when adding parameters in endpoint `api.notes.get_bbox()` (from `?` to `&`).

src/osm_easy_api/__init__.py

@@ -1,4 +1,4 @@
-VERSION = "1.1.1"
+VERSION = "2.0.0"
 
 from .data_classes import *
 from .diff import Diff, Frequency

src/osm_easy_api/api/_URLs.py

@@ -62,5 +62,6 @@ def __init__(self, base_url: str):
             "comment": six_url + "/notes/{id}/comment?text={text}",
             "close": six_url + "/notes/{id}/close",
             "reopen": six_url + "/notes/{id}/reopen",
+            "hide": six_url + "/notes/{id}",
             "search": six_url + "/notes/search?q={text}&limit={limit}&closed={closed}"
         }

src/osm_easy_api/api/api.py

@@ -58,23 +58,22 @@ def _request(self, method: _RequestMethods, url: str, auth_requirement: _Require
                 response = requests.request(str(method), url, stream=stream, data=body.encode('utf-8') if body else None, headers=headers)
         if auto_status_code_handling: assert response.status_code == 200, f"Invalid (and unexpected) response code {response.status_code} for {url}"
         return response
-
-    @staticmethod
-    def _raw_stream_parser(xml_raw_stream: "HTTPResponse") -> Generator[Tuple[str, ElementTree.Element], None, None]:
-        iterator = ElementTree.iterparse(xml_raw_stream, events=('start', 'end'))
-        for event, element in iterator:
-            yield(event, element)
-            element.clear()
 
-    def _get_generator(self, url: str, auth_requirement: _Requirement = _Requirement.OPTIONAL, auto_status_code_handling: bool = True) -> Generator[Tuple[str, ElementTree.Element], None, None] | Tuple[int, Generator[Tuple[str, ElementTree.Element], None, None]]:
+    @staticmethod
+    def _raw_stream_parser(xml_raw_stream: "HTTPResponse") -> Generator[ElementTree.Element, None, None]:
+            iterator = ElementTree.iterparse(xml_raw_stream, events=('end', ))
+            for event, element in iterator:
+                yield element
+
+    def _get_generator(self, url: str, auth_requirement: _Requirement = _Requirement.OPTIONAL, auto_status_code_handling: bool = True) -> Generator[ElementTree.Element, None, None] | Tuple[int, Generator[ElementTree.Element, None, None]]:
         response = self._request(self._RequestMethods.GET, url, auth_requirement, auto_status_code_handling=auto_status_code_handling, stream=True)
         response.raw.decode_content = True
         if auto_status_code_handling:
             return self._raw_stream_parser(response.raw)
         else:
             return (response.status_code, self._raw_stream_parser(response.raw))
 
-    def _post_generator(self, url: str, auth_requirement: _Requirement = _Requirement.OPTIONAL, auto_status_code_handling: bool = True) -> Generator[Tuple[str, ElementTree.Element], None, None] | Tuple[int, Generator[Tuple[str, ElementTree.Element], None, None]]:
+    def _post_generator(self, url: str, auth_requirement: _Requirement = _Requirement.OPTIONAL, auto_status_code_handling: bool = True) -> Generator[ElementTree.Element, None, None] | Tuple[int, Generator[ElementTree.Element, None, None]]:
         response = self._request(self._RequestMethods.POST, url, auth_requirement, auto_status_code_handling=auto_status_code_handling, stream=True)
         response.raw.decode_content = True
         if auto_status_code_handling:

src/osm_easy_api/api/endpoints/changeset.py

@@ -1,5 +1,4 @@
 from xml.dom import minidom
-from copy import copy
 
 from typing import TYPE_CHECKING, Generator, Tuple
 if TYPE_CHECKING: # pragma: no cover
@@ -22,45 +21,41 @@ def __init__(self, outer):
         self.discussion = Changeset_Discussion_Container(outer)
 
     @staticmethod
-    def _xml_to_changeset(generator: Generator[Tuple[str, 'ElementTree.Element'], None, None], include_discussion: bool = False) -> list[Changeset]:
+    def _xml_to_changesets_list(generator: Generator['ElementTree.Element', None, None], include_discussion: bool = False) -> list[Changeset]:
         """Creates Changeset instance from xml provided by API
 
         Args:
-            generator (Generator[Tuple[str, ElementTree.Element], None, None]): Generator for xml file.
+            generator (Generator[ElementTree.Element, None, None]): Generator for xml file.
             include_discussion (bool, optional): Whether xml from generator has included discussion or not. Defaults to False.
 
         Returns:
             list[Changeset]: list of Changeset objects.
         """
-        changeset_list = []
-
+        changesets_list = []
         tags = Tags()
         discussion = []
-        changeset_element = None
-        for event, element in generator:
-            if element.tag == "changeset" and event == "start":
-                tags = Tags()
-                changeset_element = copy(element)
-            elif element.tag == "tag" and event == "start":
-                tags.update({element.attrib["k"]: element.attrib["v"]})
-            elif include_discussion and element.tag == "discussion" and event == "start":
-                for comment in element:
-                    discussion.append({"date": comment.attrib["date"], "user_id": comment.attrib["uid"], "text": comment[0].text})
-            if element.tag == "changeset" and event == "end":
-                assert changeset_element, "No changeset element in API response for get changeset. Should not happen."
-                changeset_list.append(Changeset(
-                int(changeset_element.attrib["id"]),
-                changeset_element.attrib["created_at"],
-                True if changeset_element.attrib["open"] == "true" else False,
-                changeset_element.attrib["uid"],
-                changeset_element.attrib["comments_count"],
-                changeset_element.attrib["changes_count"],
-                tags,
-                discussion if include_discussion else None
-            ))
-
-        if (len(changeset_list) == 0): raise exceptions.EmptyResult()
-        return changeset_list
+        for element in generator:
+                if element.tag == "tag":
+                    tags.update({element.attrib["k"]: element.attrib["v"]})
+                elif include_discussion and element.tag == "discussion":
+                    for comment in element:
+                        discussion.append({"date": comment.attrib["date"], "user_id": comment.attrib["uid"], "text": comment[0].text})
+
+                elif element.tag == "changeset":
+                    changesets_list.append(Changeset(
+                        int(element.attrib["id"]),
+                        element.attrib["created_at"],
+                        element.attrib["open"] == "true",
+                        element.attrib["uid"],
+                        element.attrib["comments_count"],
+                        element.attrib["changes_count"],
+                        tags,
+                        discussion if include_discussion else None
+                    ))
+                    tags = Tags()
+                    element.clear()
+
+        return changesets_list
 
     def create(self, comment: str, tags: Tags | None = None) -> int:
         """Creates new changeset.
@@ -116,7 +111,7 @@ def get(self, id: int, include_discussion: bool = False) -> Changeset:
             case 404: raise exceptions.IdNotFoundError()
             case _: assert False, f"Unexpected response status code {status_code}. Please report it on github." # pragma: no cover
 
-        return self._xml_to_changeset(generator, include_discussion)[0] # type: ignore
+        return self._xml_to_changesets_list(generator, include_discussion)[0] # type: ignore
 
     def get_query(self, left: float | None = None, bottom: float | None = None, right: float | None = None, top: float | None = None,
     user_id: str | None = None, display_name: str | None = None,
@@ -175,7 +170,7 @@ def get_query(self, left: float | None = None, bottom: float | None = None, righ
             case 404: raise exceptions.IdNotFoundError()
             case _: assert False, f"Unexpected response status code {status_code}. Please report it on github." # pragma: no cover
 
-        return self._xml_to_changeset(generator) # type: ignore
+        return self._xml_to_changesets_list(generator) # type: ignore
 
     def update(self, id: int, comment: str | None = None, tags: Tags | None = None) -> Changeset:
         """Updates the changeset with new comment or tags or both.
@@ -224,7 +219,7 @@ def update(self, id: int, comment: str | None = None, tags: Tags | None = None)
             case _: assert False, f"Unexpected response status code {response.status_code}. Please report it on github." # pragma: no cover
 
         response.raw.decode_content = True
-        return self._xml_to_changeset(self.outer._raw_stream_parser(response.raw), True)[0]
+        return self._xml_to_changesets_list(self.outer._raw_stream_parser(response.raw), True)[0]
 
     def close(self, id: int) -> None:
         """Close changeset by ID.

src/osm_easy_api/api/endpoints/elements.py

@@ -1,7 +1,6 @@
 from typing import TYPE_CHECKING, TypeVar
 if TYPE_CHECKING: # pragma: no cover
     from ...api import Api
-    from xml.etree import ElementTree
 
 from copy import deepcopy
 
@@ -72,8 +71,8 @@ def get(self, element: type[Node_Way_Relation], id: int) -> Node_Way_Relation :
             case 410: raise exceptions.ElementDeleted()
             case _: assert False, f"Unexpected response status code {status_code}. Please report it on github." # pragma: no cover
 
-        for event, elem in generator:
-            if elem.tag in ("node", "way", "relation") and event == "start":
+        for elem in generator:
+            if elem.tag in ("node", "way", "relation"):
                 object = _element_to_osm_object(elem)
                 return object
 
@@ -167,8 +166,8 @@ def history(self, element: type[Node_Way_Relation], id: int) -> list[Node_Way_Re
             case _: assert False, f"Unexpected response status code {status_code}. Please report it on github." # pragma: no cover
 
         objects_list = []
-        for event, elem in generator:
-            if elem.tag == element_name and event == "start":
+        for elem in generator:
+            if elem.tag == element_name:
                 objects_list.append(_element_to_osm_object(elem))
 
         return objects_list
@@ -201,10 +200,9 @@ def version(self, element: type[Node_Way_Relation], id: int, version: int) -> No
             case 404: raise exceptions.IdNotFoundError()
             case _: assert False, f"Unexpected response status code {status_code}. Please report it on github." # pragma: no cover
 
-        for event, elem in generator:
+        for elem in generator:
             if elem.tag in ("node", "way", "relation"):
-                object = _element_to_osm_object(elem)
-                return object
+                return _element_to_osm_object(elem)
         assert False, "[ERROR::API::ENDPOINTS::ELEMENTS::version] Cannot create an element."
 
     def get_query(self, element: type[Node_Way_Relation], ids: list[int]) -> list[Node_Way_Relation]:
@@ -240,8 +238,8 @@ def get_query(self, element: type[Node_Way_Relation], ids: list[int]) -> list[No
             case _: assert False, f"Unexpected response status code {status_code}. Please report it on github." # pragma: no cover
 
         objects_list = []
-        for event, elem in generator:
-            if elem.tag == element.__name__.lower() and event == "start":
+        for elem in generator:
+            if elem.tag == element.__name__.lower():
                 objects_list.append(_element_to_osm_object(elem))
 
         return objects_list
@@ -258,14 +256,14 @@ def relations(self, element: type[Node | Way | Relation], id: int) -> list[Relat
         """
         element_name = element.__name__.lower()
         url = self.outer._url.elements["relations"].format(element_type=element_name, id=id)
-        status_code, generator = self.outer._get_generator(
+        generator = self.outer._get_generator(
             url=url,
             auth_requirement=self.outer._Requirement.NO,
-            auto_status_code_handling=False)
+            auto_status_code_handling=True)
 
         relations_list = []
-        for event, elem in generator:
-            if elem.tag == "relation" and event == "start":
+        for elem in generator:
+            if elem.tag == "relation":
                 relations_list.append(_element_to_osm_object(elem))
 
         return relations_list
@@ -280,14 +278,14 @@ def ways(self, node_id: int) -> list[Way]:
             list[Way]: List of ways.
         """
         url = self.outer._url.elements["ways"].format(id=node_id)
-        status_code, generator = self.outer._get_generator(
+        generator = self.outer._get_generator(
             url=url,
             auth_requirement=self.outer._Requirement.NO,
-            auto_status_code_handling=False)
+            auto_status_code_handling=True)
 
         ways_list = []
-        for event, elem in generator:
-            if elem.tag == "way" and event == "start":
+        for elem in generator:
+            if elem.tag == "way":
                 ways_list.append(_element_to_osm_object(elem))
 
         return ways_list
@@ -322,17 +320,16 @@ def full(self, element: type[Way_Relation], id: int) -> Way_Relation:
         nodes_dict: dict[int, Node] = {}
         ways_dict:  dict[int, Way]  = {}
         relations_dict: dict[int, Relation] = {}
-        for event, elem in generator:
-            if event == "start":
-                if elem.tag == "node":
-                    node = _element_to_osm_object(elem)
-                    nodes_dict.update({node.id: node})
-                if elem.tag == "way":
-                    way = _element_to_osm_object(elem)
-                    ways_dict.update({way.id: way})
-                if elem.tag == "relation" and element_name == "relation":
-                    relation = _element_to_osm_object(elem)
-                    relations_dict.update({relation.id: relation})
+        for elem in generator:
+            if elem.tag == "node":
+                node = _element_to_osm_object(elem)
+                nodes_dict.update({node.id: node})
+            if elem.tag == "way":
+                way = _element_to_osm_object(elem)
+                ways_dict.update({way.id: way})
+            if elem.tag == "relation" and element_name == "relation":
+                relation = _element_to_osm_object(elem)
+                relations_dict.update({relation.id: relation})
 
         for way_id in ways_dict:
             for i in range(len(ways_dict[way_id].nodes)):

src/osm_easy_api/api/endpoints/gpx.py

@@ -3,8 +3,6 @@
 if TYPE_CHECKING: # pragma: no cover
     from ...api import Api
 
-from ...api import exceptions
-
 # TODO: GPX full support and parser
 
 class Gpx_Container:

src/osm_easy_api/api/endpoints/misc.py

@@ -1,4 +1,4 @@
-from typing import TYPE_CHECKING, Generator, Tuple
+from typing import TYPE_CHECKING, Generator
 if TYPE_CHECKING: # pragma: no cover
     from xml.etree import ElementTree
     from ... import Node, Way, Relation
@@ -21,10 +21,10 @@ def versions(self) -> list:
             Returns:
                 list: List of supported versions by instance.
             """
-            gen: Generator[Tuple[str, 'ElementTree.Element'], None, None] = self.outer._get_generator(self.outer._url.misc["versions"])
+            gen: Generator['ElementTree.Element', None, None] = self.outer._get_generator(self.outer._url.misc["versions"])
             versions = []
-            for event, element in gen:
-                if element.tag == "version" and event == "start": versions.append(element.text)
+            for element in gen:
+                if element.tag == "version": versions.append(element.text)
             if len(versions) == 0: raise ValueError("[ERROR::API::MISC::versions] CAN'T FIND version")
             return versions
 
@@ -55,15 +55,16 @@ def policy_parser(dict: dict, policy_element: "ElementTree.Element") -> None:
                             dict["policy"]["imagery"]["blacklist_regex"].append(blacklist.attrib["regex"])
 
             HEAD_TAGS = ("osm", "api", "policy")
-            gen: Generator[Tuple[str, 'ElementTree.Element'], None, None] = self.outer._get_generator(self.outer._url.misc["capabilities"])
+            gen: Generator['ElementTree.Element', None, None] = self.outer._get_generator(self.outer._url.misc["capabilities"])
             return_dict = {}
 
-            for event, element in gen:
-                if element.tag in HEAD_TAGS and event == "start":
+            for element in gen:
+                if element.tag in HEAD_TAGS:
                     match element.tag:
                         case "osm": osm_parser(return_dict, element)
                         case "api": api_parser(return_dict, element)
                         case "policy": policy_parser(return_dict, element)
+                    element.clear()
 
             return return_dict
 
@@ -106,10 +107,10 @@ def permissions(self) -> list:
             Returns:
                 list: List of permissions names.
             """
-            gen: Generator[Tuple[str, 'ElementTree.Element'], None, None] = self.outer._get_generator(self.outer._url.misc["permissions"])
+            gen = self.outer._get_generator(self.outer._url.misc["permissions"])
             return_permission_list = []
 
-            for event, element in gen:
-                if element.tag == "permission" and event == "start":
+            for element in gen:
+                if element.tag == "permission":
                     return_permission_list.append(element.attrib["name"])
             return return_permission_list