hans/Nominatim
1
0
Fork 1
mirror of https://github.com/osm-search/Nominatim.git synced 2026-03-16 07:54:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

add type annotations to special phrase importer

Browse Source
This commit is contained in:
Sarah Hoffmann
2022-07-17 10:46:59 +02:00
parent 459ab3bbdc
commit 9963261d8d
9 changed files with 77 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
nominatim/tools/special_phrases/sp_importer.py
View File

@@ -13,19 +13,37 @@
The phrases already present in the database which are not
valids anymore are removed.
"""
from typing import Iterable, Tuple, Mapping, Sequence, Optional, Set
import logging
import re
from typing_extensions import Protocol
from psycopg2.sql import Identifier, SQL
from nominatim.config import Configuration
from nominatim.db.connection import Connection
from nominatim.tools.special_phrases.importer_statistics import SpecialPhrasesImporterStatistics
from nominatim.tools.special_phrases.special_phrase import SpecialPhrase
from nominatim.tokenizer.base import AbstractTokenizer
LOG = logging.getLogger()
def _classtype_table(phrase_class, phrase_type):
def _classtype_table(phrase_class: str, phrase_type: str) -> str:
""" Return the name of the table for the given class and type.
"""
return f'place_classtype_{phrase_class}_{phrase_type}'
class SpecialPhraseLoader(Protocol):
""" Protocol for classes implementing a loader for special phrases.
"""
def generate_phrases(self) -> Iterable[SpecialPhrase]:
""" Generates all special phrase terms this loader can produce.
"""
class SPImporter():
# pylint: disable-msg=too-many-instance-attributes
"""
@@ -33,21 +51,22 @@ class SPImporter():
Take a sp loader which load the phrases from an external source.
"""
def __init__(self, config, db_connection, sp_loader):
def __init__(self, config: Configuration, conn: Connection,
sp_loader: SpecialPhraseLoader) -> None:
self.config = config
self.db_connection = db_connection
self.db_connection = conn
self.sp_loader = sp_loader
self.statistics_handler = SpecialPhrasesImporterStatistics()
self.black_list, self.white_list = self._load_white_and_black_lists()
self.sanity_check_pattern = re.compile(r'^\w+$')
# This set will contain all existing phrases to be added.
# It contains tuples with the following format: (lable, class, type, operator)
self.word_phrases = set()
self.word_phrases: Set[Tuple[str, str, str, str]] = set()
# This set will contain all existing place_classtype tables which doesn't match any
# special phrases class/type on the wiki.
self.table_phrases_to_delete = set()
self.table_phrases_to_delete: Set[str] = set()
def import_phrases(self, tokenizer, should_replace):
def import_phrases(self, tokenizer: AbstractTokenizer, should_replace: bool) -> None:
"""
Iterate through all SpecialPhrases extracted from the
loader and import them into the database.
@@ -67,7 +86,7 @@ class SPImporter():
if result:
class_type_pairs.add(result)
self._create_place_classtype_table_and_indexes(class_type_pairs)
self._create_classtype_table_and_indexes(class_type_pairs)
if should_replace:
self._remove_non_existent_tables_from_db()
self.db_connection.commit()
@@ -79,7 +98,7 @@ class SPImporter():
self.statistics_handler.notify_import_done()
def _fetch_existing_place_classtype_tables(self):
def _fetch_existing_place_classtype_tables(self) -> None:
"""
Fetch existing place_classtype tables.
Fill the table_phrases_to_delete set of the class.
@@ -95,7 +114,8 @@ class SPImporter():
for row in db_cursor:
self.table_phrases_to_delete.add(row[0])
def _load_white_and_black_lists(self):
def _load_white_and_black_lists(self) \
-> Tuple[Mapping[str, Sequence[str]], Mapping[str, Sequence[str]]]:
"""
Load white and black lists from phrases-settings.json.
"""
@@ -103,7 +123,7 @@ class SPImporter():
return settings['blackList'], settings['whiteList']
def _check_sanity(self, phrase):
def _check_sanity(self, phrase: SpecialPhrase) -> bool:
"""
Check sanity of given inputs in case somebody added garbage in the wiki.
If a bad class/type is detected the system will exit with an error.
@@ -117,7 +137,7 @@ class SPImporter():
return False
return True
def _process_phrase(self, phrase):
def _process_phrase(self, phrase: SpecialPhrase) -> Optional[Tuple[str, str]]:
"""
Processes the given phrase by checking black and white list
and sanity.
@@ -145,7 +165,8 @@ class SPImporter():
return (phrase.p_class, phrase.p_type)
def _create_place_classtype_table_and_indexes(self, class_type_pairs):
def _create_classtype_table_and_indexes(self,
class_type_pairs: Iterable[Tuple[str, str]]) -> None:
"""
Create table place_classtype for each given pair.
Also create indexes on place_id and centroid.
@@ -188,7 +209,8 @@ class SPImporter():
db_cursor.execute("DROP INDEX idx_placex_classtype")
def _create_place_classtype_table(self, sql_tablespace, phrase_class, phrase_type):
def _create_place_classtype_table(self, sql_tablespace: str,
phrase_class: str, phrase_type: str) -> None:
"""
Create table place_classtype of the given phrase_class/phrase_type
if doesn't exit.
@@ -204,7 +226,8 @@ class SPImporter():
(phrase_class, phrase_type))
def _create_place_classtype_indexes(self, sql_tablespace, phrase_class, phrase_type):
def _create_place_classtype_indexes(self, sql_tablespace: str,
phrase_class: str, phrase_type: str) -> None:
"""
Create indexes on centroid and place_id for the place_classtype table.
"""
@@ -227,7 +250,7 @@ class SPImporter():
SQL(sql_tablespace)))
def _grant_access_to_webuser(self, phrase_class, phrase_type):
def _grant_access_to_webuser(self, phrase_class: str, phrase_type: str) -> None:
"""
Grant access on read to the table place_classtype for the webuser.
"""
@@ -237,7 +260,7 @@ class SPImporter():
.format(Identifier(table_name),
Identifier(self.config.DATABASE_WEBUSER)))
def _remove_non_existent_tables_from_db(self):
def _remove_non_existent_tables_from_db(self) -> None:
"""
Remove special phrases which doesn't exist on the wiki anymore.
Delete the place_classtype tables.