from nominatim.api.v1.format import RawDataList
from nominatim.api.v1 import helpers
-CONTENT_TYPE = {
- 'text': 'text/plain; charset=utf-8',
- 'xml': 'text/xml; charset=utf-8',
- 'debug': 'text/html; charset=utf-8'
-}
+CONTENT_TEXT = 'text/plain; charset=utf-8'
+CONTENT_XML = 'text/xml; charset=utf-8'
+CONTENT_HTML = 'text/html; charset=utf-8'
+CONTENT_JSON = 'application/json; charset=utf-8'
+
+CONTENT_TYPE = {'text': CONTENT_TEXT, 'xml': CONTENT_XML, 'debug': CONTENT_HTML}
class ASGIAdaptor(abc.ABC):
""" Adapter class for the different ASGI frameworks.
Wraps functionality over concrete requests and responses.
"""
- content_type: str = 'text/plain; charset=utf-8'
+ content_type: str = CONTENT_TEXT
@abc.abstractmethod
def get(self, name: str, default: Optional[str] = None) -> Optional[str]:
body of the response to 'output'.
"""
+ @abc.abstractmethod
+ def base_uri(self) -> str:
+ """ Return the URI of the original request.
+ """
+
@abc.abstractmethod
def config(self) -> Configuration:
""" Create a response from the given output. Wraps a JSONP function
around the response, if necessary.
"""
- if self.content_type == 'application/json' and status == 200:
+ if self.content_type == CONTENT_JSON and status == 200:
jsonp = self.get('json_callback')
if jsonp is not None:
if any(not part.isidentifier() for part in jsonp.split('.')):
self.raise_error('Invalid json_callback value')
output = f"{jsonp}({output})"
- self.content_type = 'application/javascript'
+ self.content_type = 'application/javascript; charset=utf-8'
return self.create_response(status, output, num_results)
message. The message will be formatted according to the
output format chosen by the request.
"""
- if self.content_type == 'text/xml; charset=utf-8':
+ if self.content_type == CONTENT_XML:
msg = f"""<?xml version="1.0" encoding="UTF-8" ?>
<error>
<code>{status}</code>
<message>{msg}</message>
</error>
"""
- elif self.content_type == 'application/json':
+ elif self.content_type == CONTENT_JSON:
msg = f"""{{"error":{{"code":{status},"message":"{msg}"}}}}"""
- elif self.content_type == 'text/html; charset=utf-8':
+ elif self.content_type == CONTENT_HTML:
loglib.log().section('Execution error')
loglib.log().var_dump('Status', status)
loglib.log().var_dump('Message', msg)
"""
if self.get_bool('debug', False):
loglib.set_log_output('html')
- self.content_type = 'text/html; charset=utf-8'
+ self.content_type = CONTENT_HTML
return True
return False
self.raise_error("Parameter 'format' must be one of: " +
', '.join(formatting.list_formats(result_type)))
- self.content_type = CONTENT_TYPE.get(fmt, 'application/json')
+ self.content_type = CONTENT_TYPE.get(fmt, CONTENT_JSON)
return fmt
result = await api.details(place,
address_details=params.get_bool('addressdetails', False),
- linked_places=params.get_bool('linkedplaces', False),
+ linked_places=params.get_bool('linkedplaces', True),
parented_places=params.get_bool('hierarchy', False),
keywords=params.get_bool('keywords', False),
geometry_output = napi.GeometryFormat.GEOJSON
if params.get_bool('polygon_geojson', False)
- else napi.GeometryFormat.NONE
+ else napi.GeometryFormat.NONE,
+ locales=locales
)
if debug:
if result is None:
params.raise_error('No place with that OSM ID found.', status=404)
- result.localize(locales)
-
output = formatting.format_result(result, fmt,
{'locales': locales,
'group_hierarchy': params.get_bool('group_hierarchy', False),
details = params.parse_geometry_details(fmt)
details['max_rank'] = helpers.zoom_to_rank(params.get_int('zoom', 18))
details['layers'] = params.get_layers()
+ details['locales'] = napi.Locales.from_accept_languages(params.get_accepted_languages())
result = await api.reverse(coord, **details)
'namedetails': params.get_bool('namedetails', False),
'addressdetails': params.get_bool('addressdetails', True)}
- if result:
- result.localize(napi.Locales.from_accept_languages(params.get_accepted_languages()))
-
output = formatting.format_result(napi.ReverseResults([result] if result else []),
fmt, fmt_options)
fmt = params.parse_format(napi.SearchResults, 'xml')
debug = params.setup_debugging()
details = params.parse_geometry_details(fmt)
+ details['locales'] = napi.Locales.from_accept_languages(params.get_accepted_languages())
places = []
for oid in (params.get('osm_ids') or '').split(','):
oid = oid.strip()
if len(oid) > 1 and oid[0] in 'RNWrnw' and oid[1:].isdigit():
- places.append(napi.OsmID(oid[0], int(oid[1:])))
+ places.append(napi.OsmID(oid[0].upper(), int(oid[1:])))
if len(places) > params.config().get_int('LOOKUP_MAX_COUNT'):
params.raise_error('Too many object IDs.')
'namedetails': params.get_bool('namedetails', False),
'addressdetails': params.get_bool('addressdetails', True)}
- results.localize(napi.Locales.from_accept_languages(params.get_accepted_languages()))
-
output = formatting.format_result(results, fmt, fmt_options)
return params.build_response(output, num_results=len(results))
helpers.feature_type_to_rank(params.get('featureType', ''))
if params.get('featureType', None) is not None:
details['layers'] = napi.DataLayer.ADDRESS
+ else:
+ details['layers'] = params.get_layers()
+
+ details['locales'] = napi.Locales.from_accept_languages(params.get_accepted_languages())
+ # unstructured query parameters
query = params.get('q', None)
+ # structured query parameters
queryparts = {}
+ for key in ('amenity', 'street', 'city', 'county', 'state', 'postalcode', 'country'):
+ details[key] = params.get(key, None)
+ if details[key]:
+ queryparts[key] = details[key]
+
try:
if query is not None:
+ if queryparts:
+ params.raise_error("Structured query parameters"
+ "(amenity, street, city, county, state, postalcode, country)"
+ " cannot be used together with 'q' parameter.")
queryparts['q'] = query
results = await _unstructured_search(query, api, details)
else:
- for key in ('amenity', 'street', 'city', 'county', 'state', 'postalcode', 'country'):
- details[key] = params.get(key, None)
- if details[key]:
- queryparts[key] = details[key]
query = ', '.join(queryparts.values())
results = await api.search_address(**details)
except UsageError as err:
params.raise_error(str(err))
- results.localize(napi.Locales.from_accept_languages(params.get_accepted_languages()))
-
if details['dedupe'] and len(results) > 1:
results = helpers.deduplicate_results(results, max_results)
(str(r.place_id) for r in results if r.place_id))
queryparts['format'] = fmt
- moreurl = urlencode(queryparts)
+ moreurl = params.base_uri() + '/search?' + urlencode(queryparts)
else:
moreurl = ''
projects / nominatim.git / blobdiff