import collections
import datetime as dt
-from nominatim_core.utils.json_writer import JsonWriter
+from ..utils.json_writer import JsonWriter
from ..status import StatusResult
from ..results import DetailedResult, ReverseResults, SearchResults, \
AddressLines, AddressLine
from ..result_formatting import FormatDispatcher
from .classtypes import ICONS
from . import format_json, format_xml
+from .. import logging as loglib
+from ..server import content_types as ct
+
class RawDataList(List[Dict[str, Any]]):
""" Data type for formatting raw data lists 'as is' in json.
"""
-dispatch = FormatDispatcher()
+
+dispatch = FormatDispatcher({'text': ct.CONTENT_TEXT,
+ 'xml': ct.CONTENT_XML,
+ 'debug': ct.CONTENT_HTML})
+
+
+@dispatch.error_format_func
+def _format_error(content_type: str, msg: str, status: int) -> str:
+ if content_type == ct.CONTENT_XML:
+ return f"""<?xml version="1.0" encoding="UTF-8" ?>
+ <error>
+ <code>{status}</code>
+ <message>{msg}</message>
+ </error>
+ """
+
+ if content_type == ct.CONTENT_JSON:
+ return f"""{{"error":{{"code":{status},"message":"{msg}"}}}}"""
+
+ if content_type == ct.CONTENT_HTML:
+ loglib.log().section('Execution error')
+ loglib.log().var_dump('Status', status)
+ loglib.log().var_dump('Message', msg)
+ return loglib.get_and_disable()
+
+ return f"ERROR {status}: {msg}"
+
@dispatch.format_func(StatusResult, 'text')
def _format_status_text(result: StatusResult, _: Mapping[str, Any]) -> str:
out = JsonWriter()
out.start_object()\
- .keyval('status', result.status)\
- .keyval('message', result.message)\
- .keyval_not_none('data_updated', result.data_updated,
- lambda v: v.isoformat())\
- .keyval('software_version', str(result.software_version))\
- .keyval_not_none('database_version', result.database_version, str)\
- .end_object()
+ .keyval('status', result.status)\
+ .keyval('message', result.message)\
+ .keyval_not_none('data_updated', result.data_updated,
+ lambda v: v.isoformat())\
+ .keyval('software_version', str(result.software_version))\
+ .keyval_not_none('database_version', result.database_version, str)\
+ .end_object()
return out()
writer.key('hierarchy').start_object()
for group, grouped in data.items():
writer.key(group).start_array()
- grouped.sort() # sorts alphabetically by local name
+ grouped.sort() # sorts alphabetically by local name
for line in grouped:
writer.raw(line).next()
writer.end_array().next()
out = JsonWriter()
out.start_object()\
- .keyval_not_none('place_id', result.place_id)\
- .keyval_not_none('parent_place_id', result.parent_place_id)
+ .keyval_not_none('place_id', result.place_id)\
+ .keyval_not_none('parent_place_id', result.parent_place_id)
if result.osm_object is not None:
out.keyval('osm_type', result.osm_object[0])\
.keyval('osm_id', result.osm_object[1])
out.keyval('category', result.category[0])\
- .keyval('type', result.category[1])\
- .keyval('admin_level', result.admin_level)\
- .keyval('localname', result.locale_name or '')\
- .keyval('names', result.names or {})\
- .keyval('addresstags', result.address or {})\
- .keyval_not_none('housenumber', result.housenumber)\
- .keyval_not_none('calculated_postcode', result.postcode)\
- .keyval_not_none('country_code', result.country_code)\
- .keyval_not_none('indexed_date', result.indexed_date, lambda v: v.isoformat())\
- .keyval_not_none('importance', result.importance)\
- .keyval('calculated_importance', result.calculated_importance())\
- .keyval('extratags', result.extratags or {})\
- .keyval_not_none('calculated_wikipedia', result.wikipedia)\
- .keyval('rank_address', result.rank_address)\
- .keyval('rank_search', result.rank_search)\
- .keyval('isarea', 'Polygon' in (geom or result.geometry.get('type') or ''))\
- .key('centroid').raw(centroid).next()\
- .key('geometry').raw(geom or centroid).next()
+ .keyval('type', result.category[1])\
+ .keyval('admin_level', result.admin_level)\
+ .keyval('localname', result.locale_name or '')\
+ .keyval('names', result.names or {})\
+ .keyval('addresstags', result.address or {})\
+ .keyval_not_none('housenumber', result.housenumber)\
+ .keyval_not_none('calculated_postcode', result.postcode)\
+ .keyval_not_none('country_code', result.country_code)\
+ .keyval_not_none('indexed_date', result.indexed_date, lambda v: v.isoformat())\
+ .keyval_not_none('importance', result.importance)\
+ .keyval('calculated_importance', result.calculated_importance())\
+ .keyval('extratags', result.extratags or {})\
+ .keyval_not_none('calculated_wikipedia', result.wikipedia)\
+ .keyval('rank_address', result.rank_address)\
+ .keyval('rank_search', result.rank_search)\
+ .keyval('isarea', 'Polygon' in (geom or result.geometry.get('type') or ''))\
+ .key('centroid').raw(centroid).next()\
+ .key('geometry').raw(geom or centroid).next()
if options.get('icon_base_url', None):
icon = ICONS.get(result.category)
extra)
-
@dispatch.format_func(SearchResults, 'geojson')
def _format_search_geojson(results: SearchResults,
- options: Mapping[str, Any]) -> str:
+ options: Mapping[str, Any]) -> str:
return format_json.format_base_geojson(results, options, False)
@dispatch.format_func(SearchResults, 'geocodejson')
def _format_search_geocodejson(results: SearchResults,
- options: Mapping[str, Any]) -> str:
+ options: Mapping[str, Any]) -> str:
return format_json.format_base_geocodejson(results, options, False)
@dispatch.format_func(SearchResults, 'json')
def _format_search_json(results: SearchResults,
- options: Mapping[str, Any]) -> str:
+ options: Mapping[str, Any]) -> str:
return format_json.format_base_json(results, options, False,
class_label='class')
@dispatch.format_func(SearchResults, 'jsonv2')
def _format_search_jsonv2(results: SearchResults,
- options: Mapping[str, Any]) -> str:
+ options: Mapping[str, Any]) -> str:
return format_json.format_base_json(results, options, False,
class_label='category')
+
@dispatch.format_func(RawDataList, 'json')
def _format_raw_data_json(results: RawDataList, _: Mapping[str, Any]) -> str:
out = JsonWriter()
out.start_object()
for k, v in res.items():
if isinstance(v, dt.datetime):
- out.keyval(k, v.isoformat(sep= ' ', timespec='seconds'))
+ out.keyval(k, v.isoformat(sep=' ', timespec='seconds'))
else:
out.keyval(k, v)
out.end_object().next()
projects / nominatim.git / blobdiff