Revert "work round typing bug in pyosmium 4.0"

[nominatim.git] / src / nominatim_db / clicmd / api.py

diff --git a/src/nominatim_db/clicmd/api.py b/src/nominatim_db/clicmd/api.py
index 20553b403dd05fabef15c9e5a344908a09489b91..dcbbb24bb4bb785c13a2389162a5c0fd1dfa96ab 100644 (file)
--- a/src/nominatim_db/clicmd/api.py
+++ b/src/nominatim_db/clicmd/api.py
@@ -7,17 +7,18 @@
 """
 Subcommand definitions for API calls from the command line.
 """
 """
 Subcommand definitions for API calls from the command line.
 """

-from typing import Dict, Any
+from typing import Dict, Any, Optional, Type, Mapping

 import argparse
 import logging
 import json
 import sys
 import argparse
 import logging
 import json
 import sys

+from functools import reduce

 
 import nominatim_api as napi
 
 import nominatim_api as napi

-import nominatim_api.v1 as api_output

 from nominatim_api.v1.helpers import zoom_to_rank, deduplicate_results
 from nominatim_api.v1.helpers import zoom_to_rank, deduplicate_results

-from nominatim_api.v1.format import dispatch as formatting
+from nominatim_api.server.content_types import CONTENT_JSON

 import nominatim_api.logging as loglib
 import nominatim_api.logging as loglib

+from ..errors import UsageError

 from .args import NominatimArgs
 
 # Do not repeat documentation of subcommand classes.
 from .args import NominatimArgs
 
 # Do not repeat documentation of subcommand classes.


@@ -42,11 +43,16 @@ EXTRADATA_PARAMS = (
     ('namedetails', 'Include a list of alternative names')
 )
 
     ('namedetails', 'Include a list of alternative names')
 )
 

+def _add_list_format(parser: argparse.ArgumentParser) -> None:
+    group = parser.add_argument_group('Other options')
+    group.add_argument('--list-formats', action='store_true',
+                       help='List supported output formats and exit.')
+
+

 def _add_api_output_arguments(parser: argparse.ArgumentParser) -> None:
 def _add_api_output_arguments(parser: argparse.ArgumentParser) -> None:

-    group = parser.add_argument_group('Output arguments')
-    group.add_argument('--format', default='jsonv2',
-                       choices=formatting.list_formats(napi.SearchResults) + ['debug'],
-                       help='Format of result')
+    group = parser.add_argument_group('Output formatting')
+    group.add_argument('--format', type=str, default='jsonv2',
+                       help='Format of result (use --list-format to see supported formats)')

     for name, desc in EXTRADATA_PARAMS:
         group.add_argument('--' + name, action='store_true', help=desc)
 
     for name, desc in EXTRADATA_PARAMS:
         group.add_argument('--' + name, action='store_true', help=desc)
 


@@ -61,6 +67,72 @@ def _add_api_output_arguments(parser: argparse.ArgumentParser) -> None:
                              "Parameter is difference tolerance in degrees."))
 
 
                              "Parameter is difference tolerance in degrees."))
 
 

+def _get_geometry_output(args: NominatimArgs) -> napi.GeometryFormat:
+    """ Get the requested geometry output format in a API-compatible
+        format.
+    """
+    if not args.polygon_output:
+        return napi.GeometryFormat.NONE
+    if args.polygon_output == 'geojson':
+        return napi.GeometryFormat.GEOJSON
+    if args.polygon_output == 'kml':
+        return napi.GeometryFormat.KML
+    if args.polygon_output == 'svg':
+        return napi.GeometryFormat.SVG
+    if args.polygon_output == 'text':
+        return napi.GeometryFormat.TEXT
+
+    try:
+        return napi.GeometryFormat[args.polygon_output.upper()]
+    except KeyError as exp:
+        raise UsageError(f"Unknown polygon output format '{args.polygon_output}'.") from exp
+
+
+def _get_locales(args: NominatimArgs, default: Optional[str]) -> napi.Locales:
+    """ Get the locales from the language parameter.
+    """
+    if args.lang:
+        return napi.Locales.from_accept_languages(args.lang)
+    if default:
+        return napi.Locales.from_accept_languages(default)
+
+    return napi.Locales()
+
+
+def _get_layers(args: NominatimArgs, default: napi.DataLayer) -> Optional[napi.DataLayer]:
+    """ Get the list of selected layers as a DataLayer enum.
+    """
+    if not args.layers:
+        return default
+
+    return reduce(napi.DataLayer.__or__,
+                  (napi.DataLayer[s.upper()] for s in args.layers))
+
+
+def _list_formats(formatter: napi.FormatDispatcher, rtype: Type[Any]) -> int:
+    for fmt in formatter.list_formats(rtype):
+        print(fmt)
+    print('debug')
+
+    return 0
+
+
+def _print_output(formatter: napi.FormatDispatcher, result: Any,
+                  fmt: str, options: Mapping[str, Any]) -> None:
+    output = formatter.format_result(result, fmt, options)
+    if formatter.get_content_type(fmt) == CONTENT_JSON:
+        # reformat the result, so it is pretty-printed
+        try:
+            json.dump(json.loads(output), sys.stdout, indent=4, ensure_ascii=False)
+        except json.decoder.JSONDecodeError as err:
+            # Catch the error here, so that data can be debugged,
+            # when people are developping custom result formatters.
+            LOG.fatal("Parsing json failed: %s\nUnformatted output:\n%s", err, output)
+    else:
+        sys.stdout.write(output)
+    sys.stdout.write('\n')
+
+

 class APISearch:
     """\
     Execute a search query.
 class APISearch:
     """\
     Execute a search query.


@@ -91,40 +163,49 @@ class APISearch:
                            help='Preferred area to find search results')
         group.add_argument('--bounded', action='store_true',
                            help='Strictly restrict results to viewbox area')
                            help='Preferred area to find search results')
         group.add_argument('--bounded', action='store_true',
                            help='Strictly restrict results to viewbox area')

-
-        group = parser.add_argument_group('Other arguments')

         group.add_argument('--no-dedupe', action='store_false', dest='dedupe',
                            help='Do not remove duplicates from the result list')
         group.add_argument('--no-dedupe', action='store_false', dest='dedupe',
                            help='Do not remove duplicates from the result list')

+        _add_list_format(parser)

 
 
     def run(self, args: NominatimArgs) -> int:
 
 
     def run(self, args: NominatimArgs) -> int:

+        formatter = napi.load_format_dispatcher('v1', args.project_dir)
+
+        if args.list_formats:
+            return _list_formats(formatter, napi.SearchResults)
+

         if args.format == 'debug':
             loglib.set_log_output('text')
         if args.format == 'debug':
             loglib.set_log_output('text')

-
-        api = napi.NominatimAPI(args.project_dir)
-
-        params: Dict[str, Any] = {'max_results': args.limit + min(args.limit, 10),
-                                  'address_details': True, # needed for display name
-                                  'geometry_output': args.get_geometry_output(),
-                                  'geometry_simplification': args.polygon_threshold,
-                                  'countries': args.countrycodes,
-                                  'excluded': args.exclude_place_ids,
-                                  'viewbox': args.viewbox,
-                                  'bounded_viewbox': args.bounded,
-                                  'locales': args.get_locales(api.config.DEFAULT_LANGUAGE)
-                                 }
-
-        if args.query:
-            results = api.search(args.query, **params)
-        else:
-            results = api.search_address(amenity=args.amenity,
-                                         street=args.street,
-                                         city=args.city,
-                                         county=args.county,
-                                         state=args.state,
-                                         postalcode=args.postalcode,
-                                         country=args.country,
-                                         **params)
+        elif not formatter.supports_format(napi.SearchResults, args.format):
+            raise UsageError(f"Unsupported format '{args.format}'. "
+                             'Use --list-formats to see supported formats.')
+
+        try:
+            with napi.NominatimAPI(args.project_dir) as api:
+                params: Dict[str, Any] = {'max_results': args.limit + min(args.limit, 10),
+                                          'address_details': True, # needed for display name
+                                          'geometry_output': _get_geometry_output(args),
+                                          'geometry_simplification': args.polygon_threshold,
+                                          'countries': args.countrycodes,
+                                          'excluded': args.exclude_place_ids,
+                                          'viewbox': args.viewbox,
+                                          'bounded_viewbox': args.bounded,
+                                          'locales': _get_locales(args, api.config.DEFAULT_LANGUAGE)
+                                         }
+
+                if args.query:
+                    results = api.search(args.query, **params)
+                else:
+                    results = api.search_address(amenity=args.amenity,
+                                                 street=args.street,
+                                                 city=args.city,
+                                                 county=args.county,
+                                                 state=args.state,
+                                                 postalcode=args.postalcode,
+                                                 country=args.country,
+                                                 **params)
+        except napi.UsageError as ex:
+            raise UsageError(ex) from ex

 
         if args.dedupe and len(results) > 1:
             results = deduplicate_results(results, args.limit)
 
         if args.dedupe and len(results) > 1:
             results = deduplicate_results(results, args.limit)


@@ -133,19 +214,10 @@ class APISearch:
             print(loglib.get_and_disable())
             return 0
 
             print(loglib.get_and_disable())
             return 0
 

-        output = api_output.format_result(
-                    results,
-                    args.format,
-                    {'extratags': args.extratags,
-                     'namedetails': args.namedetails,
-                     'addressdetails': args.addressdetails})
-        if args.format != 'xml':
-            # reformat the result, so it is pretty-printed
-            json.dump(json.loads(output), sys.stdout, indent=4, ensure_ascii=False)
-        else:
-            sys.stdout.write(output)
-        sys.stdout.write('\n')
-
+        _print_output(formatter, results, args.format,
+                      {'extratags': args.extratags,
+                       'namedetails': args.namedetails,
+                       'addressdetails': args.addressdetails})

         return 0
 
 
         return 0
 
 


@@ -161,9 +233,9 @@ class APIReverse:
 
     def add_args(self, parser: argparse.ArgumentParser) -> None:
         group = parser.add_argument_group('Query arguments')
 
     def add_args(self, parser: argparse.ArgumentParser) -> None:
         group = parser.add_argument_group('Query arguments')

-        group.add_argument('--lat', type=float, required=True,
+        group.add_argument('--lat', type=float,

                            help='Latitude of coordinate to look up (in WGS84)')
                            help='Latitude of coordinate to look up (in WGS84)')

-        group.add_argument('--lon', type=float, required=True,
+        group.add_argument('--lon', type=float,

                            help='Longitude of coordinate to look up (in WGS84)')
         group.add_argument('--zoom', type=int,
                            help='Level of detail required for the address')
                            help='Longitude of coordinate to look up (in WGS84)')
         group.add_argument('--zoom', type=int,
                            help='Level of detail required for the address')


@@ -173,39 +245,47 @@ class APIReverse:
                            help='OSM id to lookup in format <NRW><id> (may be repeated)')
 
         _add_api_output_arguments(parser)
                            help='OSM id to lookup in format <NRW><id> (may be repeated)')
 
         _add_api_output_arguments(parser)

+        _add_list_format(parser)

 
 
     def run(self, args: NominatimArgs) -> int:
 
 
     def run(self, args: NominatimArgs) -> int:

-        if args.format == 'debug':
-            loglib.set_log_output('text')
+        formatter = napi.load_format_dispatcher('v1', args.project_dir)

 
 

-        api = napi.NominatimAPI(args.project_dir)
+        if args.list_formats:
+            return _list_formats(formatter, napi.ReverseResults)

 
 

-        result = api.reverse(napi.Point(args.lon, args.lat),
-                             max_rank=zoom_to_rank(args.zoom or 18),
-                             layers=args.get_layers(napi.DataLayer.ADDRESS | napi.DataLayer.POI),
-                             address_details=True, # needed for display name
-                             geometry_output=args.get_geometry_output(),
-                             geometry_simplification=args.polygon_threshold,
-                             locales=args.get_locales(api.config.DEFAULT_LANGUAGE))
+        if args.format == 'debug':
+            loglib.set_log_output('text')
+        elif not formatter.supports_format(napi.ReverseResults, args.format):
+            raise UsageError(f"Unsupported format '{args.format}'. "
+                             'Use --list-formats to see supported formats.')
+
+        if args.lat is None or args.lon is None:
+            raise UsageError("lat' and 'lon' parameters are required.")
+
+        layers = _get_layers(args, napi.DataLayer.ADDRESS | napi.DataLayer.POI)
+
+        try:
+            with napi.NominatimAPI(args.project_dir) as api:
+                result = api.reverse(napi.Point(args.lon, args.lat),
+                                     max_rank=zoom_to_rank(args.zoom or 18),
+                                     layers=layers,
+                                     address_details=True, # needed for display name
+                                     geometry_output=_get_geometry_output(args),
+                                     geometry_simplification=args.polygon_threshold,
+                                     locales=_get_locales(args, api.config.DEFAULT_LANGUAGE))
+        except napi.UsageError as ex:
+            raise UsageError(ex) from ex

 
         if args.format == 'debug':
             print(loglib.get_and_disable())
             return 0
 
         if result:
 
         if args.format == 'debug':
             print(loglib.get_and_disable())
             return 0
 
         if result:

-            output = api_output.format_result(
-                        napi.ReverseResults([result]),
-                        args.format,
-                        {'extratags': args.extratags,
-                         'namedetails': args.namedetails,
-                         'addressdetails': args.addressdetails})
-            if args.format != 'xml':
-                # reformat the result, so it is pretty-printed
-                json.dump(json.loads(output), sys.stdout, indent=4, ensure_ascii=False)
-            else:
-                sys.stdout.write(output)
-            sys.stdout.write('\n')
+            _print_output(formatter, napi.ReverseResults([result]), args.format,
+                          {'extratags': args.extratags,
+                           'namedetails': args.namedetails,
+                           'addressdetails': args.addressdetails})

 
             return 0
 
 
             return 0
 


@@ -227,43 +307,48 @@ class APILookup:
     def add_args(self, parser: argparse.ArgumentParser) -> None:
         group = parser.add_argument_group('Query arguments')
         group.add_argument('--id', metavar='OSMID',
     def add_args(self, parser: argparse.ArgumentParser) -> None:
         group = parser.add_argument_group('Query arguments')
         group.add_argument('--id', metavar='OSMID',

-                           action='append', required=True, dest='ids',
+                           action='append', dest='ids',

                            help='OSM id to lookup in format <NRW><id> (may be repeated)')
 
         _add_api_output_arguments(parser)
                            help='OSM id to lookup in format <NRW><id> (may be repeated)')
 
         _add_api_output_arguments(parser)

+        _add_list_format(parser)

 
 
     def run(self, args: NominatimArgs) -> int:
 
 
     def run(self, args: NominatimArgs) -> int:

+        formatter = napi.load_format_dispatcher('v1', args.project_dir)
+
+        if args.list_formats:
+            return _list_formats(formatter, napi.ReverseResults)
+

         if args.format == 'debug':
             loglib.set_log_output('text')
         if args.format == 'debug':
             loglib.set_log_output('text')

+        elif not formatter.supports_format(napi.ReverseResults, args.format):
+            raise UsageError(f"Unsupported format '{args.format}'. "
+                             'Use --list-formats to see supported formats.')
+
+        if args.ids is None:
+            raise UsageError("'id' parameter required.")

 
 

-        api = napi.NominatimAPI(args.project_dir)
+        places = [napi.OsmID(o[0], int(o[1:])) for o in args.ids]
+
+        try:
+            with napi.NominatimAPI(args.project_dir) as api:
+                results = api.lookup(places,
+                                     address_details=True, # needed for display name
+                                     geometry_output=_get_geometry_output(args),
+                                     geometry_simplification=args.polygon_threshold or 0.0,
+                                     locales=_get_locales(args, api.config.DEFAULT_LANGUAGE))
+        except napi.UsageError as ex:
+            raise UsageError(ex) from ex

 
         if args.format == 'debug':
             print(loglib.get_and_disable())
             return 0
 
 
         if args.format == 'debug':
             print(loglib.get_and_disable())
             return 0
 

-        places = [napi.OsmID(o[0], int(o[1:])) for o in args.ids]
-
-        results = api.lookup(places,
-                             address_details=True, # needed for display name
-                             geometry_output=args.get_geometry_output(),
-                             geometry_simplification=args.polygon_threshold or 0.0,
-                             locales=args.get_locales(api.config.DEFAULT_LANGUAGE))
-
-        output = api_output.format_result(
-                    results,
-                    args.format,
-                    {'extratags': args.extratags,
-                     'namedetails': args.namedetails,
-                     'addressdetails': args.addressdetails})
-        if args.format != 'xml':
-            # reformat the result, so it is pretty-printed
-            json.dump(json.loads(output), sys.stdout, indent=4, ensure_ascii=False)
-        else:
-            sys.stdout.write(output)
-        sys.stdout.write('\n')
-
+        _print_output(formatter, results, args.format,
+                      {'extratags': args.extratags,
+                       'namedetails': args.namedetails,
+                       'addressdetails': args.addressdetails})

         return 0
 
 
         return 0
 
 


@@ -279,20 +364,21 @@ class APIDetails:
 
     def add_args(self, parser: argparse.ArgumentParser) -> None:
         group = parser.add_argument_group('Query arguments')
 
     def add_args(self, parser: argparse.ArgumentParser) -> None:
         group = parser.add_argument_group('Query arguments')

-        objs = group.add_mutually_exclusive_group(required=True)
-        objs.add_argument('--node', '-n', type=int,
-                          help="Look up the OSM node with the given ID.")
-        objs.add_argument('--way', '-w', type=int,
-                          help="Look up the OSM way with the given ID.")
-        objs.add_argument('--relation', '-r', type=int,
-                          help="Look up the OSM relation with the given ID.")
-        objs.add_argument('--place_id', '-p', type=int,
-                          help='Database internal identifier of the OSM object to look up')
+        group.add_argument('--node', '-n', type=int,
+                           help="Look up the OSM node with the given ID.")
+        group.add_argument('--way', '-w', type=int,
+                           help="Look up the OSM way with the given ID.")
+        group.add_argument('--relation', '-r', type=int,
+                           help="Look up the OSM relation with the given ID.")
+        group.add_argument('--place_id', '-p', type=int,
+                           help='Database internal identifier of the OSM object to look up')

         group.add_argument('--class', dest='object_class',
                            help=("Class type to disambiguated multiple entries "
                                  "of the same object."))
 
         group = parser.add_argument_group('Output arguments')
         group.add_argument('--class', dest='object_class',
                            help=("Class type to disambiguated multiple entries "
                                  "of the same object."))
 
         group = parser.add_argument_group('Output arguments')

+        group.add_argument('--format', type=str, default='json',
+                           help='Format of result (use --list-formats to see supported formats)')

         group.add_argument('--addressdetails', action='store_true',
                            help='Include a breakdown of the address into elements')
         group.add_argument('--keywords', action='store_true',
         group.add_argument('--addressdetails', action='store_true',
                            help='Include a breakdown of the address into elements')
         group.add_argument('--keywords', action='store_true',


@@ -307,9 +393,21 @@ class APIDetails:
                            help='Include geometry of result')
         group.add_argument('--lang', '--accept-language', metavar='LANGS',
                            help='Preferred language order for presenting search results')
                            help='Include geometry of result')
         group.add_argument('--lang', '--accept-language', metavar='LANGS',
                            help='Preferred language order for presenting search results')

+        _add_list_format(parser)

 
 
     def run(self, args: NominatimArgs) -> int:
 
 
     def run(self, args: NominatimArgs) -> int:

+        formatter = napi.load_format_dispatcher('v1', args.project_dir)
+
+        if args.list_formats:
+            return _list_formats(formatter, napi.DetailedResult)
+
+        if args.format == 'debug':
+            loglib.set_log_output('text')
+        elif not formatter.supports_format(napi.DetailedResult, args.format):
+            raise UsageError(f"Unsupported format '{args.format}'. "
+                             'Use --list-formats to see supported formats.')
+

         place: napi.PlaceRef
         if args.node:
             place = napi.OsmID('N', args.node, args.object_class)
         place: napi.PlaceRef
         if args.node:
             place = napi.OsmID('N', args.node, args.object_class)


@@ -317,34 +415,35 @@ class APIDetails:
             place = napi.OsmID('W', args.way, args.object_class)
         elif args.relation:
             place = napi.OsmID('R', args.relation, args.object_class)
             place = napi.OsmID('W', args.way, args.object_class)
         elif args.relation:
             place = napi.OsmID('R', args.relation, args.object_class)

-        else:
-            assert args.place_id is not None
+        elif  args.place_id is not None:

             place = napi.PlaceID(args.place_id)
             place = napi.PlaceID(args.place_id)

+        else:
+            raise UsageError('One of the arguments --node/-n --way/-w '
+                             '--relation/-r --place_id/-p is required/')
+
+        try:
+            with napi.NominatimAPI(args.project_dir) as api:
+                locales = _get_locales(args, api.config.DEFAULT_LANGUAGE)
+                result = api.details(place,
+                                     address_details=args.addressdetails,
+                                     linked_places=args.linkedplaces,
+                                     parented_places=args.hierarchy,
+                                     keywords=args.keywords,
+                                     geometry_output=napi.GeometryFormat.GEOJSON
+                                                     if args.polygon_geojson
+                                                     else napi.GeometryFormat.NONE,
+                                    locales=locales)
+        except napi.UsageError as ex:
+            raise UsageError(ex) from ex

 
 

-        api = napi.NominatimAPI(args.project_dir)
-
-        locales = args.get_locales(api.config.DEFAULT_LANGUAGE)
-        result = api.details(place,
-                             address_details=args.addressdetails,
-                             linked_places=args.linkedplaces,
-                             parented_places=args.hierarchy,
-                             keywords=args.keywords,
-                             geometry_output=napi.GeometryFormat.GEOJSON
-                                             if args.polygon_geojson
-                                             else napi.GeometryFormat.NONE,
-                            locales=locales)
-
+        if args.format == 'debug':
+            print(loglib.get_and_disable())
+            return 0

 
         if result:
 
         if result:

-            output = api_output.format_result(
-                        result,
-                        'json',
-                        {'locales': locales,
-                         'group_hierarchy': args.group_hierarchy})
-            # reformat the result, so it is pretty-printed
-            json.dump(json.loads(output), sys.stdout, indent=4, ensure_ascii=False)
-            sys.stdout.write('\n')
-
+            _print_output(formatter, result, args.format or 'json',
+                          {'locales': locales,
+                           'group_hierarchy': args.group_hierarchy})

             return 0
 
         LOG.error("Object not found in database.")
             return 0
 
         LOG.error("Object not found in database.")


@@ -362,13 +461,34 @@ class APIStatus:
     """
 
     def add_args(self, parser: argparse.ArgumentParser) -> None:
     """
 
     def add_args(self, parser: argparse.ArgumentParser) -> None:

-        formats = api_output.list_formats(napi.StatusResult)

         group = parser.add_argument_group('API parameters')
         group = parser.add_argument_group('API parameters')

-        group.add_argument('--format', default=formats[0], choices=formats,
-                           help='Format of result')
+        group.add_argument('--format', type=str, default='text',
+                           help='Format of result (use --list-formats to see supported formats)')
+        _add_list_format(parser)

 
 
     def run(self, args: NominatimArgs) -> int:
 
 
     def run(self, args: NominatimArgs) -> int:

-        status = napi.NominatimAPI(args.project_dir).status()
-        print(api_output.format_result(status, args.format, {}))
+        formatter = napi.load_format_dispatcher('v1', args.project_dir)
+
+        if args.list_formats:
+            return _list_formats(formatter, napi.StatusResult)
+
+        if args.format == 'debug':
+            loglib.set_log_output('text')
+        elif not formatter.supports_format(napi.StatusResult, args.format):
+            raise UsageError(f"Unsupported format '{args.format}'. "
+                             'Use --list-formats to see supported formats.')
+
+        try:
+            with napi.NominatimAPI(args.project_dir) as api:
+                status = api.status()
+        except napi.UsageError as ex:
+            raise UsageError(ex) from ex
+
+        if args.format == 'debug':
+            print(loglib.get_and_disable())
+            return 0
+
+        _print_output(formatter, status, args.format, {})
+

         return 0
         return 0