nominatim/db/sqlalchemy_types/key_value.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) 2023 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 A custom type that implements a simple key-value store of strings.
   9 """
  10 from typing import Any
  11
  12 import sqlalchemy as sa
  13 from sqlalchemy.dialects.postgresql import HSTORE
  14 from sqlalchemy.dialects.sqlite import JSON as sqlite_json
  15
  16 from nominatim.typing import SaDialect, SaColumn
  17
  18 # pylint: disable=all
  19
  20 class KeyValueStore(sa.types.TypeDecorator[Any]):
  21     """ Dialect-independent type of a simple key-value store of strings.
  22     """
  23     impl = HSTORE
  24     cache_ok = True
  25
  26     def load_dialect_impl(self, dialect: SaDialect) -> sa.types.TypeEngine[Any]:
  27         if dialect.name == 'postgresql':
  28             return HSTORE() # type: ignore[no-untyped-call]
  29
  30         return sqlite_json(none_as_null=True)
  31
  32
  33     class comparator_factory(sa.types.UserDefinedType.Comparator): # type: ignore[type-arg]
  34
  35         def merge(self, other: SaColumn) -> 'sa.Operators':
  36             """ Merge the values from the given KeyValueStore into this
  37                 one, overwriting values where necessary. When the argument
  38                 is null, nothing happens.
  39             """
  40             return self.op('||')(sa.func.coalesce(other,
  41                                                   sa.type_coerce('', KeyValueStore)))
  42
  43
  44         def has_key(self, key: SaColumn) -> 'sa.Operators':
  45             """ Return true if the key is cotained in the store.
  46             """
  47             return self.op('?', is_comparison=True)(key)