What3Words API v3 (#444)

geopy/geocoders/__init__.py

@@ -212,6 +212,7 @@
     "LiveAddress",
     "TomTom",
     "What3Words",
+    "What3WordsV3",
     "Yandex",
 )
 
@@ -243,6 +244,7 @@
 from geopy.geocoders.smartystreets import LiveAddress
 from geopy.geocoders.tomtom import TomTom
 from geopy.geocoders.what3words import What3Words
+from geopy.geocoders.what3wordsv3 import What3WordsV3
 from geopy.geocoders.yandex import Yandex
 
 SERVICE_TO_GEOCODER = {
@@ -273,6 +275,7 @@
     "liveaddress": LiveAddress,
     "tomtom": TomTom,
     "what3words": What3Words,
+    "what3wordsv3": What3WordsV3,
     "yandex": Yandex,
 }
 

geopy/geocoders/what3wordsv3.py

@@ -0,0 +1,219 @@
+import re
+from functools import partial
+from urllib.parse import urlencode
+
+from geopy import exc
+from geopy.geocoders.base import DEFAULT_SENTINEL, Geocoder
+from geopy.location import Location
+from geopy.util import logger
+
+__all__ = ("What3WordsV3", )
+
+
+class What3WordsV3(Geocoder):
+    """What3Words geocoder.
+
+    Documentation at:
+        https://developer.what3words.com/public-api/
+    """
+
+    multiple_word_re = re.compile(
+        r"[^\W\d\_]+\.{1,1}[^\W\d\_]+\.{1,1}[^\W\d\_]+$", re.U
+    )
+
+    geocode_path = '/v3/convert-to-coordinates'
+    reverse_path = '/v3/convert-to-3wa'
+
+    def __init__(
+            self,
+            api_key,
+            *,
+            timeout=DEFAULT_SENTINEL,
+            proxies=DEFAULT_SENTINEL,
+            user_agent=None,
+            ssl_context=DEFAULT_SENTINEL,
+            adapter_factory=None
+    ):
+        """
+
+        :param str api_key: Key provided by What3Words
+            (https://accounts.what3words.com/register).
+
+        :param int timeout:
+            See :attr:`geopy.geocoders.options.default_timeout`.
+
+        :param dict proxies:
+            See :attr:`geopy.geocoders.options.default_proxies`.
+
+        :param str user_agent:
+            See :attr:`geopy.geocoders.options.default_user_agent`.
+
+        :type ssl_context: :class:`ssl.SSLContext`
+        :param ssl_context:
+            See :attr:`geopy.geocoders.options.default_ssl_context`.
+
+        :param callable adapter_factory:
+            See :attr:`geopy.geocoders.options.default_adapter_factory`.
+
+            .. versionadded:: 2.0
+        """
+        super().__init__(
+            scheme='https',
+            timeout=timeout,
+            proxies=proxies,
+            user_agent=user_agent,
+            ssl_context=ssl_context,
+            adapter_factory=adapter_factory,
+        )
+
+        self.api_key = api_key
+        domain = 'api.what3words.com'
+        self.geocode_api = '%s://%s%s' % (self.scheme, domain, self.geocode_path)
+        self.reverse_api = '%s://%s%s' % (self.scheme, domain, self.reverse_path)
+
+    def _check_query(self, query):
+        """
+        Check query validity with regex
+        """
+        if not self.multiple_word_re.match(query):
+            return False
+        else:
+            return True
+
+    def geocode(
+            self,
+            query,
+            *,
+            exactly_one=True,
+            timeout=DEFAULT_SENTINEL
+    ):
+
+        """
+        Return a location point for a `3 words` query. If the `3 words` address
+        doesn't exist, a :class:`geopy.exc.GeocoderQueryError` exception will be
+        thrown.
+
+        :param str query: The 3-word address you wish to geocode.
+
+        :param bool exactly_one: Return one result or a list of results, if
+            available. Due to the address scheme there is always exactly one
+            result for each `3 words` address, so this parameter is rather
+            useless for this geocoder.
+
+        :param int timeout: Time, in seconds, to wait for the geocoding service
+            to respond before raising a :class:`geopy.exc.GeocoderTimedOut`
+            exception. Set this only if you wish to override, on this call
+            only, the value set during the geocoder's initialization.
+
+        :rtype: :class:`geopy.location.Location` or a list of them, if
+            ``exactly_one=False``.
+        """
+
+        if not self._check_query(query):
+            raise exc.GeocoderQueryError(
+                "Search string must be 'word.word.word'"
+            )
+
+        params = {
+            'words': query,
+            'key': self.api_key,
+        }
+
+        url = "?".join((self.geocode_api, urlencode(params)))
+        logger.debug("%s.geocode: %s", self.__class__.__name__, url)
+        callback = partial(self._parse_json, exactly_one=exactly_one)
+        return self._call_geocoder(url, callback, timeout=timeout)
+
+    def _parse_json(self, resources, exactly_one=True):
+        """
+        Parse type, words, latitude, and longitude and language from a
+        JSON response.
+        """
+
+        error = resources.get('error', None)
+
+        if error is not None:
+            # https://developer.what3words.com/public-api/docs#error-handling
+            exc_msg = "Error returned by What3Words: %s" % resources["error"]["message"]
+            exc_code = error.get('code')
+            if exc_code in ['MissingKey', 'InvalidKey']:
+                raise exc.GeocoderAuthenticationFailure(exc_msg)
+
+            raise exc.GeocoderQueryError(exc_msg)
+
+        def parse_resource(resource):
+            """
+            Parse record.
+            """
+
+            if 'coordinates' in resource:
+                words = resource['words']
+                position = resource['coordinates']
+                latitude, longitude = position['lat'], position['lng']
+                if latitude and longitude:
+                    latitude = float(latitude)
+                    longitude = float(longitude)
+
+                return Location(words, (latitude, longitude), resource)
+            else:
+                raise exc.GeocoderParseError('Error parsing result.')
+
+        location = parse_resource(resources)
+        if exactly_one:
+            return location
+        else:
+            return [location]
+
+    def reverse(
+            self,
+            query,
+            *,
+            lang='en',
+            exactly_one=True,
+            timeout=DEFAULT_SENTINEL
+    ):
+        """
+        Return a `3 words` address by location point. Each point on surface has
+        a `3 words` address, so there's always a non-empty response.
+
+        :param query: The coordinates for which you wish to obtain the 3 word
+            address.
+        :type query: :class:`geopy.point.Point`, list or tuple of ``(latitude,
+            longitude)``, or string as ``"%(latitude)s, %(longitude)s"``.
+
+        :param str lang: two character language codes as supported by the
+            API (https://docs.what3words.com/api/v2/#lang).
+
+        :param bool exactly_one: Return one result or a list of results, if
+            available. Due to the address scheme there is always exactly one
+            result for each `3 words` address, so this parameter is rather
+            useless for this geocoder.
+
+        :param int timeout: Time, in seconds, to wait for the geocoding service
+            to respond before raising a :class:`geopy.exc.GeocoderTimedOut`
+            exception. Set this only if you wish to override, on this call
+            only, the value set during the geocoder's initialization.
+
+        :rtype: :class:`geopy.location.Location` or a list of them, if
+            ``exactly_one=False``.
+
+        """
+        lang = lang.lower()
+
+        params = {
+            'coordinates': self._coerce_point_to_string(query),
+            'language': lang.lower(),
+            'key': self.api_key,
+        }
+
+        url = "?".join((self.reverse_api, urlencode(params)))
+
+        logger.debug("%s.reverse: %s", self.__class__.__name__, url)
+        callback = partial(self._parse_reverse_json, exactly_one=exactly_one)
+        return self._call_geocoder(url, callback, timeout=timeout)
+
+    def _parse_reverse_json(self, resources, exactly_one=True):
+        """
+        Parses a location from a single-result reverse API call.
+        """
+        return self._parse_json(resources, exactly_one)

test/geocoders/what3wordsv3.py

@@ -0,0 +1,96 @@
+from unittest.mock import patch
+
+import pytest
+
+import geopy.exc
+import geopy.geocoders
+from geopy.geocoders import What3WordsV3
+from test.geocoders.util import BaseTestGeocoder, env
+
+
+class TestUnitWhat3WordsV3:
+    dummy_api_key = 'DUMMYKEY1234'
+
+    async def test_user_agent_custom(self):
+        geocoder = What3WordsV3(
+            api_key=self.dummy_api_key,
+            user_agent='my_user_agent/1.0'
+        )
+        assert geocoder.headers['User-Agent'] == 'my_user_agent/1.0'
+
+    @patch.object(geopy.geocoders.options, 'default_scheme', 'http')
+    def test_default_scheme_is_ignored(self):
+        geocoder = What3WordsV3(api_key=self.dummy_api_key)
+        assert geocoder.scheme == 'https'
+
+
+@pytest.mark.skipif(
+    not bool(env.get('WHAT3WORDS_KEY')),
+    reason="No WHAT3WORDS_KEY env variable set"
+)
+class TestWhat3WordsV3(BaseTestGeocoder):
+    @classmethod
+    def make_geocoder(cls, **kwargs):
+        return What3WordsV3(
+            env['WHAT3WORDS_KEY'],
+            timeout=3,
+            **kwargs
+        )
+
+    async def test_geocode(self):
+        await self.geocode_run(
+            {"query": "piped.gains.jangle"},
+            {"latitude": 53.037611, "longitude": 11.565012},
+        )
+
+    async def test_reverse(self):
+        await self.reverse_run(
+            {"query": "53.037611,11.565012", "lang": 'DE'},
+            {"address": 'fortschrittliche.voll.schnitt'},
+        )
+
+    async def test_unicode_query(self):
+        await self.geocode_run(
+            {
+                "query": (
+                    "\u0070\u0069\u0070\u0065\u0064\u002e\u0067"
+                    "\u0061\u0069\u006e\u0073\u002e\u006a\u0061"
+                    "\u006e\u0067\u006c\u0065"
+                )
+            },
+            {"latitude": 53.037611, "longitude": 11.565012},
+        )
+
+    async def test_empty_response(self):
+        with pytest.raises(geopy.exc.GeocoderQueryError):
+            await self.geocode_run(
+                {"query": "definitely.not.existingiswearrrr"},
+                {},
+                expect_failure=True
+            )
+
+    async def test_not_exactly_one(self):
+        await self.geocode_run(
+            {"query": "piped.gains.jangle", "exactly_one": False},
+            {"latitude": 53.037611, "longitude": 11.565012},
+        )
+        await self.reverse_run(
+            {"query": (53.037611, 11.565012), "exactly_one": False},
+            {"address": "piped.gains.jangle"},
+        )
+
+    async def test_result_language(self):
+        await self.reverse_run(
+            {"query": (53.037611, 11.565012), "lang": "en", "exactly_one": False},
+            {"address": "piped.gains.jangle"},
+        )
+
+    async def test_check_query(self):
+        result_check_threeword_query = self.geocoder._check_query(
+            "\u0066\u0061\u0068\u0072\u0070\u0072"
+            "\u0065\u0069\u0073\u002e\u006c\u00fc"
+            "\u0067\u006e\u0065\u0072\u002e\u006b"
+            "\u0075\u0074\u0073\u0063\u0068\u0065"
+        )
+
+        assert result_check_threeword_query