from textwrap import dedent
from ..config import Configuration
-from ..db.connection import connect, Connection, server_version_tuple,\
+from ..db.connection import connect, Connection, server_version_tuple, \
index_exists, table_exists, execute_scalar
from ..db import properties
from ..errors import UsageError
CHECKLIST = []
+
class CheckState(Enum):
""" Possible states of a check. FATAL stops check execution entirely.
"""
NOT_APPLICABLE = 3
WARN = 4
+
CheckResult = Union[CheckState, Tuple[CheckState, Mapping[str, Any]]]
CheckFunc = Callable[[Connection, Configuration], CheckResult]
+
def _check(hint: Optional[str] = None) -> Callable[[CheckFunc], CheckFunc]:
""" Decorator for checks. It adds the function to the list of
checks to execute and adds the code for printing progress messages.
return decorator
+
class _BadConnection:
def __init__(self, msg: str) -> None:
""" Dummy function to provide the implementation.
"""
+
def check_database(config: Configuration) -> int:
""" Run a number of checks on the database and return the status.
"""
try:
conn = connect(config.get_libpq_dsn())
except UsageError as err:
- conn = _BadConnection(str(err)) # type: ignore[assignment]
+ conn = _BadConnection(str(err)) # type: ignore[assignment]
overall_result = 0
for check in CHECKLIST:
'idx_osmline_parent_osm_id',
'idx_postcode_id',
'idx_postcode_postcode'
- ]
+ ]
# These won't exist if --reverse-only import was used
if table_exists(conn, 'search_name'):
return CheckState.OK
+
@_check(hint="""\
Database version ({db_version}) doesn't match Nominatim version ({nom_version})
instruction=instruction,
config=config)
+
@_check(hint="""\
placex table not found
return CheckState.OK
if freeze.is_frozen(conn):
- index_cmd="""\
+ index_cmd = """\
Database is marked frozen, it cannot be updated.
Low counts of unindexed places are fine."""
return CheckState.WARN, dict(count=cnt, index_cmd=index_cmd)