fix: 705 Status Code Documentation

[nominatim.git] / nominatim / tools / replication.py

diff --git a/nominatim/tools/replication.py b/nominatim/tools/replication.py
index fab3d2db57dc54822d4178261718a17ca132024e..db706bf67563d795b863da9b0d025412b8554c0f 100644 (file)
--- a/nominatim/tools/replication.py
+++ b/nominatim/tools/replication.py
@@ -7,6 +7,7 @@
 """
 Functions for updating a database from a replication source.
 """
 """
 Functions for updating a database from a replication source.
 """

+from typing import ContextManager, MutableMapping, Any, Generator, cast

 from contextlib import contextmanager
 import datetime as dt
 from enum import Enum
 from contextlib import contextmanager
 import datetime as dt
 from enum import Enum


@@ -14,6 +15,7 @@ import logging
 import time
 
 from nominatim.db import status
 import time
 
 from nominatim.db import status

+from nominatim.db.connection import Connection

 from nominatim.tools.exec_utils import run_osm2pgsql
 from nominatim.errors import UsageError
 
 from nominatim.tools.exec_utils import run_osm2pgsql
 from nominatim.errors import UsageError
 


@@ -27,7 +29,7 @@ except ImportError as exc:
 
 LOG = logging.getLogger()
 
 
 LOG = logging.getLogger()
 

-def init_replication(conn, base_url):
+def init_replication(conn: Connection, base_url: str) -> None:

     """ Set up replication for the server at the given base URL.
     """
     LOG.info("Using replication source: %s", base_url)
     """ Set up replication for the server at the given base URL.
     """
     LOG.info("Using replication source: %s", base_url)


@@ -51,7 +53,7 @@ def init_replication(conn, base_url):
     LOG.warning("Updates initialised at sequence %s (%s)", seq, date)
 
 
     LOG.warning("Updates initialised at sequence %s (%s)", seq, date)
 
 

-def check_for_updates(conn, base_url):
+def check_for_updates(conn: Connection, base_url: str) -> int:

     """ Check if new data is available from the replication service at the
         given base URL.
     """
     """ Check if new data is available from the replication service at the
         given base URL.
     """


@@ -84,7 +86,7 @@ class UpdateState(Enum):
     NO_CHANGES = 3
 
 
     NO_CHANGES = 3
 
 

-def update(conn, options):
+def update(conn: Connection, options: MutableMapping[str, Any]) -> UpdateState:

     """ Update database from the next batch of data. Returns the state of
         updates according to `UpdateState`.
     """
     """ Update database from the next batch of data. Returns the state of
         updates according to `UpdateState`.
     """


@@ -95,6 +97,8 @@ def update(conn, options):
                   "Please run 'nominatim replication --init' first.")
         raise UsageError("Replication not set up.")
 
                   "Please run 'nominatim replication --init' first.")
         raise UsageError("Replication not set up.")
 

+    assert startdate is not None
+

     if not indexed and options['indexed_only']:
         LOG.info("Skipping update. There is data that needs indexing.")
         return UpdateState.MORE_PENDING
     if not indexed and options['indexed_only']:
         LOG.info("Skipping update. There is data that needs indexing.")
         return UpdateState.MORE_PENDING


@@ -132,17 +136,17 @@ def update(conn, options):
     return UpdateState.UP_TO_DATE
 
 
     return UpdateState.UP_TO_DATE
 
 

-def _make_replication_server(url):
+def _make_replication_server(url: str) -> ContextManager[ReplicationServer]:

     """ Returns a ReplicationServer in form of a context manager.
 
         Creates a light wrapper around older versions of pyosmium that did
         not support the context manager interface.
     """
     if hasattr(ReplicationServer, '__enter__'):
     """ Returns a ReplicationServer in form of a context manager.
 
         Creates a light wrapper around older versions of pyosmium that did
         not support the context manager interface.
     """
     if hasattr(ReplicationServer, '__enter__'):

-        return ReplicationServer(url)
+        return cast(ContextManager[ReplicationServer], ReplicationServer(url))

 
     @contextmanager
 
     @contextmanager

-    def get_cm():
+    def get_cm() -> Generator[ReplicationServer, None, None]:

         yield ReplicationServer(url)
 
     return get_cm()
         yield ReplicationServer(url)
 
     return get_cm()