src/nominatim_db/clicmd/convert.py

   1 # SPDX-License-Identifier: GPL-3.0-or-later
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2024 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Implementation of the 'convert' subcommand.
   9 """
  10 from typing import Set, Any, Union, Optional, Sequence
  11 import argparse
  12 import asyncio
  13 from pathlib import Path
  14
  15 from nominatim_core.errors import UsageError
  16 from .args import NominatimArgs
  17
  18 # Do not repeat documentation of subcommand classes.
  19 # pylint: disable=C0111
  20 # Using non-top-level imports to avoid eventually unused imports.
  21 # pylint: disable=E0012,C0415
  22
  23 class WithAction(argparse.Action):
  24     """ Special action that saves a list of flags, given on the command-line
  25         as `--with-foo` or `--without-foo`.
  26     """
  27     def __init__(self, option_strings: Sequence[str], dest: Any,
  28                  default: bool = True, **kwargs: Any) -> None:
  29         if 'nargs' in kwargs:
  30             raise ValueError("nargs not allowed.")
  31         if option_strings is None:
  32             raise ValueError("Positional parameter not allowed.")
  33
  34         self.dest_set = kwargs.pop('dest_set')
  35         full_option_strings = []
  36         for opt in option_strings:
  37             if not opt.startswith('--'):
  38                 raise ValueError("short-form options not allowed")
  39             if default:
  40                 self.dest_set.add(opt[2:])
  41             full_option_strings.append(f"--with-{opt[2:]}")
  42             full_option_strings.append(f"--without-{opt[2:]}")
  43
  44         super().__init__(full_option_strings, argparse.SUPPRESS, nargs=0, **kwargs)
  45
  46
  47     def __call__(self, parser: argparse.ArgumentParser, namespace: argparse.Namespace,
  48                  values: Union[str, Sequence[Any], None],
  49                  option_string: Optional[str] = None) -> None:
  50         assert option_string
  51         if option_string.startswith('--with-'):
  52             self.dest_set.add(option_string[7:])
  53         if option_string.startswith('--without-'):
  54             self.dest_set.discard(option_string[10:])
  55
  56
  57 class ConvertDB:
  58     """ Convert an existing database into a different format. (EXPERIMENTAL)
  59
  60         Dump a read-only version of the database in a different format.
  61         At the moment only a SQLite database suitable for reverse lookup
  62         can be created.
  63     """
  64
  65     def __init__(self) -> None:
  66         self.options: Set[str] = set()
  67
  68     def add_args(self, parser: argparse.ArgumentParser) -> None:
  69         parser.add_argument('--format', default='sqlite',
  70                             choices=('sqlite', ),
  71                             help='Format of the output database (must be sqlite currently)')
  72         parser.add_argument('--output', '-o', required=True, type=Path,
  73                             help='File to write the database to.')
  74         group = parser.add_argument_group('Switches to define database layout'
  75                                           '(currently no effect)')
  76         group.add_argument('--reverse', action=WithAction, dest_set=self.options, default=True,
  77                            help='Enable/disable support for reverse and lookup API'
  78                                 ' (default: enabled)')
  79         group.add_argument('--search', action=WithAction, dest_set=self.options, default=True,
  80                            help='Enable/disable support for search API (default: disabled)')
  81         group.add_argument('--details', action=WithAction, dest_set=self.options, default=True,
  82                            help='Enable/disable support for details API (default: enabled)')
  83
  84
  85     def run(self, args: NominatimArgs) -> int:
  86         if args.output.exists():
  87             raise UsageError(f"File '{args.output}' already exists. Refusing to overwrite.")
  88
  89         if args.format == 'sqlite':
  90             from ..tools import convert_sqlite
  91
  92             asyncio.run(convert_sqlite.convert(args.project_dir, args.output, self.options))
  93             return 0
  94
  95         return 1