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 ..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