split SearchResult type

Use adapted types for the different result types. This makes it
easier to have adapted output formatting and means there are only
result fields that are filled.
This commit is contained in:
Sarah Hoffmann
2023-03-14 14:21:35 +01:00
parent d03fd3f883
commit 00e3a752c9
10 changed files with 274 additions and 148 deletions

View File

@@ -11,7 +11,7 @@ Data classes are part of the public API while the functions are for
internal use only. That's why they are implemented as free-standing functions
instead of member functions.
"""
from typing import Optional, Tuple, Dict, Sequence
from typing import Optional, Tuple, Dict, Sequence, TypeVar, Type
import enum
import dataclasses
import datetime as dt
@@ -69,16 +69,15 @@ WordInfos = Sequence[WordInfo]
@dataclasses.dataclass
class SearchResult:
""" Data class collecting all available information about a search result.
class BaseResult:
""" Data class collecting information common to all
types of search results.
"""
source_table: SourceTable
category: Tuple[str, str]
centroid: Point
place_id : Optional[int] = None
parent_place_id: Optional[int] = None
linked_place_id: Optional[int] = None
osm_object: Optional[Tuple[str, int]] = None
admin_level: int = 15
@@ -96,8 +95,6 @@ class SearchResult:
country_code: Optional[str] = None
indexed_date: Optional[dt.datetime] = None
address_rows: Optional[AddressLines] = None
linked_rows: Optional[AddressLines] = None
parented_rows: Optional[AddressLines] = None
@@ -106,10 +103,6 @@ class SearchResult:
geometry: Dict[str, str] = dataclasses.field(default_factory=dict)
def __post_init__(self) -> None:
if self.indexed_date is not None and self.indexed_date.tzinfo is None:
self.indexed_date = self.indexed_date.replace(tzinfo=dt.timezone.utc)
@property
def lat(self) -> float:
""" Get the latitude (or y) of the center point of the place.
@@ -131,93 +124,138 @@ class SearchResult:
"""
return self.importance or (0.7500001 - (self.rank_search/40.0))
BaseResultT = TypeVar('BaseResultT', bound=BaseResult)
@dataclasses.dataclass
class DetailedResult(BaseResult):
""" A search result with more internal information from the database
added.
"""
parent_place_id: Optional[int] = None
linked_place_id: Optional[int] = None
indexed_date: Optional[dt.datetime] = None
def _filter_geometries(row: SaRow) -> Dict[str, str]:
return {k[9:]: v for k, v in row._mapping.items() # pylint: disable=W0212
if k.startswith('geometry_')}
def create_from_placex_row(row: SaRow) -> SearchResult:
""" Construct a new SearchResult and add the data from the result row
from the placex table.
def create_from_placex_row(row: Optional[SaRow],
class_type: Type[BaseResultT]) -> Optional[BaseResultT]:
""" Construct a new result and add the data from the result row
from the placex table. 'class_type' defines the type of result
to return. Returns None if the row is None.
"""
return SearchResult(source_table=SourceTable.PLACEX,
place_id=row.place_id,
parent_place_id=row.parent_place_id,
linked_place_id=row.linked_place_id,
osm_object=(row.osm_type, row.osm_id),
category=(row.class_, row.type),
admin_level=row.admin_level,
names=row.name,
address=row.address,
extratags=row.extratags,
housenumber=row.housenumber,
postcode=row.postcode,
wikipedia=row.wikipedia,
rank_address=row.rank_address,
rank_search=row.rank_search,
importance=row.importance,
country_code=row.country_code,
indexed_date=getattr(row, 'indexed_date'),
centroid=Point.from_wkb(row.centroid.data),
geometry=_filter_geometries(row))
if row is None:
return None
return class_type(source_table=SourceTable.PLACEX,
place_id=row.place_id,
osm_object=(row.osm_type, row.osm_id),
category=(row.class_, row.type),
admin_level=row.admin_level,
names=row.name,
address=row.address,
extratags=row.extratags,
housenumber=row.housenumber,
postcode=row.postcode,
wikipedia=row.wikipedia,
rank_address=row.rank_address,
rank_search=row.rank_search,
importance=row.importance,
country_code=row.country_code,
centroid=Point.from_wkb(row.centroid.data),
geometry=_filter_geometries(row))
def create_from_osmline_row(row: SaRow) -> SearchResult:
""" Construct a new SearchResult and add the data from the result row
from the osmline table.
def create_from_osmline_row(row: Optional[SaRow],
class_type: Type[BaseResultT]) -> Optional[BaseResultT]:
""" Construct a new result and add the data from the result row
from the address interpolation table osmline. 'class_type' defines
the type of result to return. Returns None if the row is None.
If the row contains a housenumber, then the housenumber is filled out.
Otherwise the result contains the interpolation information in extratags.
"""
return SearchResult(source_table=SourceTable.OSMLINE,
place_id=row.place_id,
parent_place_id=row.parent_place_id,
osm_object=('W', row.osm_id),
category=('place', 'houses'),
address=row.address,
postcode=row.postcode,
extratags={'startnumber': str(row.startnumber),
'endnumber': str(row.endnumber),
'step': str(row.step)},
country_code=row.country_code,
indexed_date=getattr(row, 'indexed_date'),
centroid=Point.from_wkb(row.centroid.data),
geometry=_filter_geometries(row))
if row is None:
return None
hnr = getattr(row, 'housenumber', None)
res = class_type(source_table=SourceTable.OSMLINE,
place_id=row.place_id,
osm_object=('W', row.osm_id),
category=('place', 'houses' if hnr is None else 'house'),
address=row.address,
postcode=row.postcode,
country_code=row.country_code,
centroid=Point.from_wkb(row.centroid.data),
geometry=_filter_geometries(row))
if hnr is None:
res.extratags = {'startnumber': str(row.startnumber),
'endnumber': str(row.endnumber),
'step': str(row.step)}
else:
res.housenumber = str(hnr)
return res
def create_from_tiger_row(row: SaRow) -> SearchResult:
""" Construct a new SearchResult and add the data from the result row
from the Tiger table.
def create_from_tiger_row(row: Optional[SaRow],
class_type: Type[BaseResultT]) -> Optional[BaseResultT]:
""" Construct a new result and add the data from the result row
from the Tiger data interpolation table. 'class_type' defines
the type of result to return. Returns None if the row is None.
If the row contains a housenumber, then the housenumber is filled out.
Otherwise the result contains the interpolation information in extratags.
"""
return SearchResult(source_table=SourceTable.TIGER,
place_id=row.place_id,
parent_place_id=row.parent_place_id,
category=('place', 'houses'),
postcode=row.postcode,
extratags={'startnumber': str(row.startnumber),
'endnumber': str(row.endnumber),
'step': str(row.step)},
country_code='us',
centroid=Point.from_wkb(row.centroid.data),
geometry=_filter_geometries(row))
if row is None:
return None
hnr = getattr(row, 'housenumber', None)
res = class_type(source_table=SourceTable.TIGER,
place_id=row.place_id,
category=('place', 'houses' if hnr is None else 'house'),
postcode=row.postcode,
country_code='us',
centroid=Point.from_wkb(row.centroid.data),
geometry=_filter_geometries(row))
if hnr is None:
res.extratags = {'startnumber': str(row.startnumber),
'endnumber': str(row.endnumber),
'step': str(row.step)}
else:
res.housenumber = str(hnr)
return res
def create_from_postcode_row(row: SaRow) -> SearchResult:
""" Construct a new SearchResult and add the data from the result row
from the postcode centroid table.
def create_from_postcode_row(row: Optional[SaRow],
class_type: Type[BaseResultT]) -> Optional[BaseResultT]:
""" Construct a new result and add the data from the result row
from the postcode table. 'class_type' defines
the type of result to return. Returns None if the row is None.
"""
return SearchResult(source_table=SourceTable.POSTCODE,
place_id=row.place_id,
parent_place_id=row.parent_place_id,
category=('place', 'postcode'),
names={'ref': row.postcode},
rank_search=row.rank_search,
rank_address=row.rank_address,
country_code=row.country_code,
centroid=Point.from_wkb(row.centroid.data),
indexed_date=row.indexed_date,
geometry=_filter_geometries(row))
if row is None:
return None
return class_type(source_table=SourceTable.POSTCODE,
place_id=row.place_id,
category=('place', 'postcode'),
names={'ref': row.postcode},
rank_search=row.rank_search,
rank_address=row.rank_address,
country_code=row.country_code,
centroid=Point.from_wkb(row.centroid.data),
geometry=_filter_geometries(row))
async def add_result_details(conn: SearchConnection, result: SearchResult,
async def add_result_details(conn: SearchConnection, result: BaseResult,
details: LookupDetails) -> None:
""" Retrieve more details from the database according to the
parameters specified in 'details'.
@@ -262,7 +300,7 @@ def _result_row_to_address_row(row: SaRow) -> AddressLine:
distance=row.distance)
async def complete_address_details(conn: SearchConnection, result: SearchResult) -> None:
async def complete_address_details(conn: SearchConnection, result: BaseResult) -> None:
""" Retrieve information about places that make up the address of the result.
"""
housenumber = -1
@@ -292,6 +330,7 @@ async def complete_address_details(conn: SearchConnection, result: SearchResult)
for row in await conn.execute(sql):
result.address_rows.append(_result_row_to_address_row(row))
# pylint: disable=consider-using-f-string
def _placex_select_address_row(conn: SearchConnection,
centroid: Point) -> SaSelect:
@@ -308,7 +347,7 @@ def _placex_select_address_row(conn: SearchConnection,
""" % centroid).label('distance'))
async def complete_linked_places(conn: SearchConnection, result: SearchResult) -> None:
async def complete_linked_places(conn: SearchConnection, result: BaseResult) -> None:
""" Retrieve information about places that link to the result.
"""
result.linked_rows = []
@@ -322,7 +361,7 @@ async def complete_linked_places(conn: SearchConnection, result: SearchResult) -
result.linked_rows.append(_result_row_to_address_row(row))
async def complete_keywords(conn: SearchConnection, result: SearchResult) -> None:
async def complete_keywords(conn: SearchConnection, result: BaseResult) -> None:
""" Retrieve information about the search terms used for this place.
"""
t = conn.t.search_name
@@ -342,7 +381,7 @@ async def complete_keywords(conn: SearchConnection, result: SearchResult) -> Non
result.address_keywords.append(WordInfo(*row))
async def complete_parented_places(conn: SearchConnection, result: SearchResult) -> None:
async def complete_parented_places(conn: SearchConnection, result: BaseResult) -> None:
""" Retrieve information about places that the result provides the
address for.
"""