2 Functions for updating a database from a replication source.
9 from osmium.replication.server import ReplicationServer
10 from osmium import WriteHandler
12 from ..db import status
13 from .exec_utils import run_osm2pgsql
15 LOG = logging.getLogger()
17 def init_replication(conn, base_url):
18 """ Set up replication for the server at the given base URL.
20 LOG.info("Using replication source: %s", base_url)
21 date = status.compute_database_date(conn)
23 # margin of error to make sure we get all data
24 date -= dt.timedelta(hours=3)
26 repl = ReplicationServer(base_url)
28 seq = repl.timestamp_to_sequence(date)
31 LOG.fatal("Cannot reach the configured replication service '%s'.\n"
32 "Does the URL point to a directory containing OSM update data?",
34 raise RuntimeError("Failed to reach replication service")
36 status.set_status(conn, date=date, seq=seq)
38 LOG.warning("Updates intialised at sequence %s (%s)", seq, date)
41 def check_for_updates(conn, base_url):
42 """ Check if new data is available from the replication service at the
45 _, seq, _ = status.get_status(conn)
48 LOG.error("Replication not set up. "
49 "Please run 'nominatim replication --init' first.")
52 state = ReplicationServer(base_url).get_state_info()
55 LOG.error("Cannot get state for URL %s.", base_url)
58 if state.sequence <= seq:
59 LOG.warning("Database is up to date.")
62 LOG.warning("New data available (%i => %i).", seq, state.sequence)
65 class UpdateState(Enum):
66 """ Possible states after an update has run.
74 def update(conn, options):
75 """ Update database from the next batch of data. Returns the state of
76 updates according to `UpdateState`.
78 startdate, startseq, indexed = status.get_status(conn)
81 LOG.error("Replication not set up. "
82 "Please run 'nominatim replication --init' first.")
83 raise RuntimeError("Replication not set up.")
85 if not indexed and options['indexed_only']:
86 LOG.info("Skipping update. There is data that needs indexing.")
87 return UpdateState.MORE_PENDING
89 last_since_update = dt.datetime.now(dt.timezone.utc) - startdate
90 update_interval = dt.timedelta(seconds=options['update_interval'])
91 if last_since_update < update_interval:
92 duration = (update_interval - last_since_update).seconds
93 LOG.warning("Sleeping for %s sec before next update.", duration)
96 if options['import_file'].exists():
97 options['import_file'].unlink()
99 # Read updates into file.
100 repl = ReplicationServer(options['base_url'])
102 outhandler = WriteHandler(str(options['import_file']))
103 endseq = repl.apply_diffs(outhandler, startseq,
104 max_size=options['max_diff_size'] * 1024)
108 return UpdateState.NO_CHANGES
110 # Consume updates with osm2pgsql.
111 options['append'] = True
112 run_osm2pgsql(options)
114 # Write the current status to the file
115 endstate = repl.get_state_info(endseq)
116 status.set_status(conn, endstate.timestamp, seq=endseq, indexed=False)
118 return UpdateState.UP_TO_DATE