src/nominatim_api/utils/json_writer.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 Streaming JSON encoder.
   9 """
  10 from typing import Any, TypeVar, Optional, Callable
  11 import io
  12 try:
  13     import ujson as json
  14 except ModuleNotFoundError:
  15     import json # type: ignore[no-redef]
  16
  17 T = TypeVar('T') # pylint: disable=invalid-name
  18
  19 class JsonWriter:
  20     """ JSON encoder that renders the output directly into an output
  21         stream. This is a very simple writer which produces JSON in a
  22         compact as possible form.
  23
  24         The writer does not check for syntactic correctness. It is the
  25         responsibility of the caller to call the write functions in an
  26         order that produces correct JSON.
  27
  28         All functions return the writer object itself so that function
  29         calls can be chained.
  30     """
  31
  32     def __init__(self) -> None:
  33         self.data = io.StringIO()
  34         self.pending = ''
  35
  36
  37     def __call__(self) -> str:
  38         """ Return the rendered JSON content as a string.
  39             The writer remains usable after calling this function.
  40         """
  41         if self.pending:
  42             assert self.pending in (']', '}')
  43             self.data.write(self.pending)
  44             self.pending = ''
  45         return self.data.getvalue()
  46
  47
  48     def start_object(self) -> 'JsonWriter':
  49         """ Write the open bracket of a JSON object.
  50         """
  51         if self.pending:
  52             self.data.write(self.pending)
  53         self.pending = '{'
  54         return self
  55
  56
  57     def end_object(self) -> 'JsonWriter':
  58         """ Write the closing bracket of a JSON object.
  59         """
  60         assert self.pending in (',', '{', '')
  61         if self.pending == '{':
  62             self.data.write(self.pending)
  63         self.pending = '}'
  64         return self
  65
  66
  67     def start_array(self) -> 'JsonWriter':
  68         """ Write the opening bracket of a JSON array.
  69         """
  70         if self.pending:
  71             self.data.write(self.pending)
  72         self.pending = '['
  73         return self
  74
  75
  76     def end_array(self) -> 'JsonWriter':
  77         """ Write the closing bracket of a JSON array.
  78         """
  79         assert self.pending in (',', '[', ']', ')', '')
  80         if self.pending not in (',', ''):
  81             self.data.write(self.pending)
  82         self.pending = ']'
  83         return self
  84
  85
  86     def key(self, name: str) -> 'JsonWriter':
  87         """ Write the key string of a JSON object.
  88         """
  89         assert self.pending
  90         self.data.write(self.pending)
  91         self.data.write(json.dumps(name, ensure_ascii=False))
  92         self.pending = ':'
  93         return self
  94
  95
  96     def value(self, value: Any) -> 'JsonWriter':
  97         """ Write out a value as JSON. The function uses the json.dumps()
  98             function for encoding the JSON. Thus any value that can be
  99             encoded by that function is permissible here.
 100         """
 101         return self.raw(json.dumps(value, ensure_ascii=False))
 102
 103
 104     def float(self, value: float, precision: int) -> 'JsonWriter':
 105         """ Write out a float value with the given precision.
 106         """
 107         return self.raw(f"{value:0.{precision}f}")
 108
 109     def next(self) -> 'JsonWriter':
 110         """ Write out a delimiter comma between JSON object or array elements.
 111         """
 112         if self.pending:
 113             self.data.write(self.pending)
 114         self.pending = ','
 115         return self
 116
 117
 118     def raw(self, raw_json: str) -> 'JsonWriter':
 119         """ Write out the given value as is. This function is useful if
 120             a value is already available in JSON format.
 121         """
 122         if self.pending:
 123             self.data.write(self.pending)
 124             self.pending = ''
 125         self.data.write(raw_json)
 126         return self
 127
 128
 129     def keyval(self, key: str, value: Any) -> 'JsonWriter':
 130         """ Write out an object element with the given key and value.
 131             This is a shortcut for calling 'key()', 'value()' and 'next()'.
 132         """
 133         self.key(key)
 134         self.value(value)
 135         return self.next()
 136
 137
 138     def keyval_not_none(self, key: str, value: Optional[T],
 139                         transform: Optional[Callable[[T], Any]] = None) -> 'JsonWriter':
 140         """ Write out an object element only if the value is not None.
 141             If 'transform' is given, it must be a function that takes the
 142             value type and returns a JSON encodable type. The transform
 143             function will be called before the value is written out.
 144         """
 145         if value is not None:
 146             self.key(key)
 147             self.value(transform(value) if transform else value)
 148             self.next()
 149         return self