safety-screen/Nominatim
1
0
Fork 0
forked from hans/Nominatim
Code Pull Requests Activity

Removed magic mocking, using monkeypatch instead, and using a placex table to simulate a 'real database'

Browse Source
This commit is contained in:
anqixxx
2025-04-11 12:03:57 -07:00
parent 1a323165f9
commit 1952290359
2 changed files with 106 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
src/nominatim_db/tools/special_phrases/sp_importer.py
View File

@@ -16,7 +16,7 @@
from typing import Iterable, Tuple, Mapping, Sequence, Optional, Set
import logging
import re
import json
import json
from psycopg.sql import Identifier, SQL
from ...typing import Protocol
@@ -65,37 +65,37 @@ class SPImporter():
# special phrases class/type on the wiki.
self.table_phrases_to_delete: Set[str] = set()
def get_classtype_pairs_style(self) -> Set[Tuple[str, str]]:
def get_classtype_pairs_style(self) -> Set[Tuple[str, str]]:
"""
Returns list of allowed special phrases from the the style file,
restricting to a list of combinations of classes and types
Returns list of allowed special phrases from the the style file,
restricting to a list of combinations of classes and types
which have a 'main' property
Note: This requirement was from 2021 and I am a bit unsure if it is still relevant
"""
style_file = self.config.get_import_style_file() # this gives the path, so i will import it as a json
style_file = self.config.get_import_style_file() # import style file as json
with open(style_file, 'r') as file:
style_data = json.loads(f'[{file.read()}]')
style_combinations = set()
for _map in style_data: # following ../settings/import-extratags.style
for _map in style_data: # following ../settings/import-extratags.style
classes = _map.get("keys", [])
values = _map.get("values", {})
for _type, properties in values.items():
if "main" in properties and _type: # make sure the tag is not an empty string. since type is the value of the main tag
if "main" in properties and _type: # make sure the tag is a non-empty string
for _class in classes:
style_combinations.add((_class, _type))
style_combinations.add((_class, _type)) # type is the value of the main tag
return style_combinations
def get_classtype_pairs(self) -> Set[Tuple[str, str]]:
def get_classtype_pairs(self) -> Set[Tuple[str, str]]:
"""
Returns list of allowed special phrases from the database,
restricting to a list of combinations of classes and types
whic occur more than 100 times
Returns list of allowed special phrases from the database,
restricting to a list of combinations of classes and types
which occur more than 100 times
"""
db_combinations = set()
db_combinations = set()
query = """
SELECT class AS CLS, type AS typ
FROM placex
@@ -104,13 +104,12 @@ class SPImporter():
"""
with self.db_connection.cursor() as db_cursor:
db_cursor.execute(SQL(query))
db_cursor.execute(SQL(query))
for row in db_cursor.fetchall():
db_combinations.add((row[0], row[1]))
return db_combinations
return db_combinations
def import_phrases(self, tokenizer: AbstractTokenizer, should_replace: bool) -> None:
"""
Iterate through all SpecialPhrases extracted from the
@@ -131,11 +130,10 @@ class SPImporter():
if result:
class_type_pairs.add(result)
self._create_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()
with tokenizer.name_analyzer() as analyzer:
@@ -235,7 +233,7 @@ class SPImporter():
LOG.warning("Skipping phrase %s=%s: not in allowed special phrases",
phrase_class, phrase_type)
continue
table_name = _classtype_table(phrase_class, phrase_type)
if table_name in self.table_phrases_to_delete:
@@ -268,13 +266,13 @@ class SPImporter():
"""
table_name = _classtype_table(phrase_class, phrase_type)
with self.db_connection.cursor() as db_cursor:
db_cursor.execute(SQL("""CREATE TABLE IF NOT EXISTS {} {} AS
SELECT place_id AS place_id,
st_centroid(geometry) AS centroid
FROM placex
WHERE class = %s AND type = %s
""").format(Identifier(table_name), SQL(sql_tablespace)),
(phrase_class, phrase_type))
db_cursor.execute(SQL(
"""CREATE TABLE IF NOT EXISTS {} {} AS
SELECT place_id AS place_id,
st_centroid(geometry) AS centroid
FROM placex WHERE class = %s AND type = %s
""").format(Identifier(table_name), SQL(sql_tablespace)),
(phrase_class, phrase_type))
def _create_place_classtype_indexes(self, sql_tablespace: str,
phrase_class: str, phrase_type: str) -> None:
@@ -321,4 +319,3 @@ class SPImporter():
drop_tables(self.db_connection, *self.table_phrases_to_delete)
for _ in self.table_phrases_to_delete:
self.statistics_handler.notify_one_table_deleted()