osm2pgsql-replication

#!/usr/bin/env python3

# SPDX-License-Identifier: GPL-2.0-or-later
#
# This file is part of osm2pgsql (https://osm2pgsql.org/).
#
# Copyright (C) 2006-2024 by the osm2pgsql developer community.
# For a full list of authors see the git log.

"""
Update an osm2pgsql database with changes from a OSM replication server.

This tool initialises the updating process by looking at the import file
or the newest object in the database. The state is then saved in a table
in the database. Subsequent runs download newly available data and apply
it to the database.

See the help of the `init` and `update` command for more information on
how to use %(prog)s.
"""

from argparse import ArgumentParser, RawDescriptionHelpFormatter
import datetime as dt
import json
import logging
import sys
import subprocess
import tempfile
from textwrap import dedent
from pathlib import Path
import urllib.request as urlrequest

missing_modules = []

try:
    import psycopg2 as psycopg
    from psycopg2 import sql
except ImportError:
    try:
        import psycopg
        from psycopg import sql
    except ImportError:
        missing_modules.append('psycopg2')

try:
    from osmium.replication.server import ReplicationServer
    from osmium.replication.utils import get_replication_header
    from osmium import WriteHandler
except ImportError:
    missing_modules.append('osmium')

LOG = logging.getLogger()

OSM2PGSQL_PATH = Path(__file__).parent.resolve() / 'osm2pgsql'

def pretty_format_timedelta(seconds):
    seconds = int(seconds)
    (minutes, seconds) = divmod(seconds, 60)
    (hours, minutes) = divmod(minutes, 60)
    (days, hours) = divmod(hours, 24)
    (weeks, days) = divmod(days, 7)

    output = []
    # If weeks > 1 but hours == 0, we still want to show "0 hours"
    if weeks > 0:
        output.append("{} week(s)".format(weeks))
    if days > 0 or weeks > 0:
        output.append("{} day(s)".format(days))
    if hours > 0 or days > 0 or weeks > 0:
        output.append("{} hour(s)".format(hours))
    if minutes > 0 or hours > 0 or days > 0 or weeks > 0:
        output.append("{} minute(s)".format(minutes))

    output.append("{} second(s)".format(seconds))

    output = " ".join(output)
    return output


def osm_date(date):
    return date.strftime('%Y-%m-%dT%H:%M:%SZ')


def from_osm_date(datestr):
    return dt.datetime.strptime(datestr, '%Y-%m-%dT%H:%M:%SZ').replace(tzinfo=dt.timezone.utc)


def start_point(param):
    if param.isdigit():
        return int(param)

    if sys.version_info >= (3, 7):
        try:
            date = dt.datetime.fromisoformat(param)
            if date.tzinfo is None:
                date = date.replace(tzinfo=dt.timezone.utc)
            return date
        except ValueError:
            pass

    return from_osm_date(param)


class DBError(Exception):

    def __init__(self, errno, msg):
        self.errno = errno
        self.msg = msg


class DBConnection:

    def __init__(self, args):
        self.schema = args.middle_schema

        # If dbname looks like a conninfo string use it as such
        if args.database and any(part in args.database for part in ['=', '://']):
            self.conn = psycopg.connect(args.database,
                                        fallback_application_name="osm2pgsql-replication")
        else:
            self.conn = psycopg.connect(dbname=args.database, user=args.username,
                                        host=args.host, port=args.port,
                                        fallback_application_name="osm2pgsql-replication")

        self.name = self.conn.info.dbname

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        if self.conn is not None:
            self.conn.close()

    def table_exists(self, table_name):
        with self.conn.cursor() as cur:
            cur.execute('SELECT * FROM pg_tables WHERE tablename = %s and schemaname = %s ',
                        (table_name, self.schema))
            return cur.rowcount > 0

    def table_id(self, name):
        return sql.Identifier(self.schema, name)


class Osm2pgsqlProperties:
    PROP_TABLE_NAME = 'osm2pgsql_properties'

    def __init__(self, db):
        self.db = db
        self.is_updatable = self._get_prop('updatable') == 'true'

    def _get_prop(self, name):
        with self.db.conn.cursor() as cur:
            cur.execute(sql.SQL("SELECT value FROM {} WHERE property = %s")
                           .format(self.db.table_id(self.PROP_TABLE_NAME)),
                        (name, ))
            return cur.fetchone()[0] if cur.rowcount == 1 else None

    def _set_prop(self, name, value):
        with self.db.conn.cursor() as cur:
            cur.execute(sql.SQL("""INSERT INTO {} (property, value) VALUES (%s, %s)
                                   ON CONFLICT (property)
                                   DO UPDATE SET value = EXCLUDED.value""")
                           .format(self.db.table_id(self.PROP_TABLE_NAME)),
                        (name, value))

    def get_replication_base(self, server, start_at):
        seq, date = None, None
        if server is None:
            server = self._get_prop('replication_base_url')
            if server:
                seq = self._get_prop('replication_sequence_number')
                date = self._get_prop('replication_timestamp')
                if date is not None:
                    date = from_osm_date(date)
            else:
                server = 'https://planet.openstreetmap.org/replication/minute'

        if isinstance(start_at, dt.datetime):
            return server, None, start_at

        if seq is None or isinstance(start_at, int):
            date = self._get_prop('current_timestamp')
            if date is None:
                raise DBError(1, "Cannot get timestamp from database. "
                                 "Use --start-at to set an explicit date.")

            date = from_osm_date(date)
            date -= dt.timedelta(minutes=start_at or 180)
            seq = None
        else:
            seq = int(seq)

        return server, seq, date

    def get_replication_state(self):
        if not self.db.table_exists(self.PROP_TABLE_NAME):
            raise DBError(1, "Cannot find replication status table. Run 'osm2pgsql-replication init' first.")

        base_url = self._get_prop('replication_base_url')
        seq = self._get_prop('replication_sequence_number')
        date = self._get_prop('replication_timestamp')

        if base_url is None or seq is None or date is None:
            raise DBError(2, "Updates not set up correctly. Run 'osm2pgsql-replication init' first.")

        return base_url, int(seq), from_osm_date(date)

    def write_replication_state(self, base_url, seq, date):
        self._set_prop('replication_base_url', base_url)
        self._set_prop('replication_sequence_number', seq)
        self._set_prop('replication_timestamp', osm_date(date))
        self.db.conn.commit()


class LegacyProperties:

    def __init__(self, db, prefix):
        self.db = db
        self.prop_table = f'{prefix}_replication_status'
        self.way_table = f'{prefix}_ways'
        self.is_updatable = db.table_exists(self.way_table)

    def get_replication_base(self, server, start_at):
        """ Determine the date of the database from the newest object in the
            database.
        """
        if server is None:
            server = 'https://planet.openstreetmap.org/replication/minute'

        if isinstance(start_at, dt.datetime):
            return server, None, start_at

        # First, find the way with the highest ID in the database
        # Using nodes would be more reliable but those are not cached by osm2pgsql.
        with self.db.conn.cursor() as cur:
            cur.execute(sql.SQL("SELECT max(id) FROM {}")
                           .format(self.db.table_id(self.way_table)))
            osmid = cur.fetchone()[0] if cur.rowcount == 1 else None

            if osmid is None:
                raise DBError(1, "No data found in the database.")

        LOG.debug("Using way id %d for timestamp lookup", osmid)
        # Get the way from the API to find the timestamp when it was created.
        url = 'https://www.openstreetmap.org/api/0.6/way/{}/1'.format(osmid)
        headers = {"User-Agent" : "osm2pgsql-replication",
                   "Accept" : "application/json"}
        with urlrequest.urlopen(urlrequest.Request(url, headers=headers)) as response:
            data = json.loads(response.read().decode('utf-8'))

        if not data.get('elements') or not 'timestamp' in data['elements'][0]:
            raise DBError(1, "The way data downloaded from the API does not contain valid data.\n"
                             f"URL used: {url}")

        date = data['elements'][0]['timestamp']
        LOG.debug("Found timestamp %s", date)

        try:
            date = from_osm_date(date)
        except ValueError:
            raise DBError(1, f"Cannot parse timestamp '{date}'")

        date -= dt.timedelta(minutes=start_at or 180)

        return server, None, date

    def get_replication_state(self):
        if not self.db.table_exists(self.prop_table):
            raise DBError(1, "Cannot find replication status table. Run 'osm2pgsql-replication init' first.")

        with self.db.conn.cursor() as cur:
            cur.execute(sql.SQL('SELECT * FROM {}').format(self.db.table_id(self.prop_table)))
            if cur.rowcount != 1:
                raise DBError(2, "Updates not set up correctly. Run 'osm2pgsql-replication init' first.")

            base_url, seq, date = cur.fetchone()

        if base_url is None or seq is None or date is None:
            raise DBError(2, "Updates not set up correctly. Run 'osm2pgsql-replication init' first.")

        return base_url, seq, date

    def write_replication_state(self, base_url, seq, date):
        table = self.db.table_id(self.prop_table)
        with self.db.conn.cursor() as cur:
            if not self.db.table_exists(self.prop_table):
                cur.execute(sql.SQL("""CREATE TABLE {}
                                       (url TEXT,
                                        sequence INTEGER,
                                        importdate TIMESTAMP WITH TIME ZONE)
                                    """).format(table))
            cur.execute(sql.SQL('TRUNCATE {}').format(table))
            if date:
                cur.execute(sql.SQL('INSERT INTO {} VALUES(%s, %s, %s)').format(table),
                            (base_url, seq, date))
            else:
                cur.execute(sql.SQL('INSERT INTO {} VALUES(%s, %s)').format(table),
                            (base_url, seq))
        self.db.conn.commit()


def get_status_info(props, args):
    results = {'status': 0}

    base_url, db_seq, db_ts = props.get_replication_state()

    db_ts = db_ts.astimezone(dt.timezone.utc)
    results['server'] = {'base_url': base_url}
    results['local'] = {'sequence': db_seq, 'timestamp': osm_date(db_ts)}

    repl = ReplicationServer(base_url)
    state_info = repl.get_state_info()
    if state_info is None:
        # PyOsmium was unable to download the state information
        results['status'] = 3
        results['error'] = "Unable to download the state information from {}".format(base_url)
    else:
        results['status'] = 0
        now = dt.datetime.now(dt.timezone.utc)

        server_seq, server_ts = state_info
        server_ts = server_ts.astimezone(dt.timezone.utc)

        results['server']['sequence'] = server_seq
        results['server']['timestamp'] = osm_date(server_ts)
        results['server']['age_sec'] = int((now-server_ts).total_seconds())

        results['local']['age_sec'] = int((now - db_ts).total_seconds())

    return results


def status(props, args):
    """\
    Print information about the current replication status, optionally as JSON.

    Sample output:

        2021-08-17 15:20:28 [INFO]: Using replication service 'https://planet.openstreetmap.org/replication/minute', which is at sequence 4675115 ( 2021-08-17T13:19:43Z )
        2021-08-17 15:20:28 [INFO]: Replication server's most recent data is <1 minute old
        2021-08-17 15:20:28 [INFO]: Local database is 8288 sequences behind the server, i.e. 5 day(s) 20 hour(s) 58 minute(s)
        2021-08-17 15:20:28 [INFO]: Local database's most recent data is 5 day(s) 20 hour(s) 59 minute(s) old


    With the `--json` option, the status is printed as a json object.

        {
          "server": {
            "base_url": "https://planet.openstreetmap.org/replication/minute",
            "sequence": 4675116,
            "timestamp": "2021-08-17T13:20:43Z",
            "age_sec": 27
          },
          "local": {
            "sequence": 4666827,
            "timestamp": "2021-08-11T16:21:09Z",
            "age_sec": 507601
          },
          "status": 0
        }


    `status` is 0 if there were no problems getting the status. 1 & 2 for
    improperly set up replication. 3 for network issues. If status ≠ 0, then
    the `error` key is an error message (as string). `status` is used as the
    exit code.

    `server` is the replication server's current status. `sequence` is it's
    sequence number, `timestamp` the time of that, and 'age_sec' the age of the
    data in seconds.

    `local` is the status of your server.
    """
    try:
        results = get_status_info(props, args)
    except DBError as err:
        results = {'status': err.errno, 'error': err.msg}

    if args.json:
        print(json.dumps(results))
    else:
        if results['status'] != 0:
            LOG.fatal(results['error'])
        else:
            print("Using replication service '{}', which is at sequence {} ( {} )".format(
                     results['server']['base_url'], results['server']['sequence'], results['server']['timestamp']))
            print("Replication server's most recent data is {} old".format(pretty_format_timedelta(results['server']['age_sec'])))

            if results['local']['sequence'] == results['server']['sequence']:
                print("Local database is up to date with server")
            else:
                print("Local database is {} sequences behind the server, i.e. {}".format(
                        results['server']['sequence'] - results['local']['sequence'],
                        pretty_format_timedelta(results['local']['age_sec'] - results['server']['age_sec'])
                    ))

            print("Local database's most recent data is {} old".format(pretty_format_timedelta(results['local']['age_sec'])))

    return results['status']


def init(props, args):
    """\
    Initialise the replication process.

    This function sets the replication service to use and determines from
    which date to apply updates. You must call this function at least once
    to set up the replication process. It can safely be called again later
    to change the replication servers or to roll back the update process and
    reapply updates.

    There are different methods available for initialisation. When no
    additional parameters are given, the data is initialised from the data
    in the database. If the data was imported from a file with replication
    information and the properties table is available (for osm2pgsql >= 1.9)
    then the replication from the file is used. Otherwise the minutely
    update service from openstreetmap.org is used as the default replication
    service. The start date is either taken from the database timestamp
    (for osm2pgsql >= 1.9) or determined from the newest way in the database
    by querying the OSM API about its creation date.

    The replication service can be changed with the `--server` parameter.
    To use a different start date, add `--start-at` with an absolute
    ISO timestamp (e.g. 2007-08-20T12:21:53Z). When the program determines the
    start date from the database timestamp or way creation date, then it
    subtracts another 3 hours by default to ensure that all new changes are
    available. To change this rollback period, use `--start-at` with the
    number of minutes to rollback. This rollback mode can also be used to
    force initialisation to use the database date and ignore the date
    from the replication information in the file.

    The initialisation process can also use replication information from
    an OSM file directly and ignore all other date information.
    Use the command `%(prog)s --osm-file <filename>` for this.
    """
    if args.osm_file is None:
        base_url, seq, date = props.get_replication_base(args.server, args.start_at)
    else:
        base_url, seq, date = get_replication_header(args.osm_file)
        if base_url is None or (seq is None and date is None):
            raise DBError(1, f"File '{args.osm_file}' has no usable replication headers. Use '--server' instead.")

    repl = ReplicationServer(base_url)
    if seq is None:
        seq = repl.timestamp_to_sequence(date)
        if seq is None:
            raise DBError(1, f"Cannot reach the configured replication service '{base_url}'.\n"
                             "Does the URL point to a directory containing OSM update data?")

    if date is None:
        state = repl.get_state_info(seq)
        if state is None:
            raise DBError(1, f"Cannot reach the configured replication service '{base_url}'.\n"
                             "Does the URL point to a directory containing OSM update data?")
        date = from_osm_date(state.timestamp)

    props.write_replication_state(base_url, seq, date)

    LOG.info("Initialised updates for service '%s'.", base_url)
    LOG.info("Starting at sequence %d (%s).", seq, osm_date(date))

    return 0


def update(props, args):
    """\
    Download newly available data and apply it to the database.

    The data is downloaded in chunks of `--max-diff-size` MB. Each chunk is
    saved in a temporary file and imported with osm2pgsql from there. The
    temporary file is normally deleted afterwards unless you state an explicit
    location with `--diff-file`. Once the database is up to date with the
    replication source, the update process exits with 0.

    Any additional arguments to osm2pgsql need to be given after `--`. Database
    and the prefix parameter are handed through to osm2pgsql. They do not need
    to be repeated. `--append` and `--slim` will always be added as well.

    Use the `--post-processing` parameter to execute a script after osm2pgsql has
    run successfully. If the updates consists of multiple runs because the
    maximum size of downloaded data was reached, then the script is executed
    each time that osm2pgsql has run. When the post-processing fails, then
    the entire update run is considered a failure and the replication information
    is not updated. That means that when 'update' is run the next time it will
    recommence with downloading the diffs again and reapplying them to the
    database. This is usually safe. The script receives two parameters:
    the sequence ID and timestamp of the last successful run. The timestamp
    may be missing in the rare case that the replication service stops responding
    after the updates have been downloaded.
    """
    base_url, seq, ts = props.get_replication_state()

    initial_local_timestamp = ts
    LOG.info("Using replication service '%s'.", base_url)
    local_db_age_sec = int((dt.datetime.now(dt.timezone.utc) - ts).total_seconds())

    repl = ReplicationServer(base_url)
    current = repl.get_state_info()
    if current is None:
        raise DBError(1, f"Cannot reach the configured replication service '{base_url}'.\n"
                         "Does the URL point to a directory containing OSM update data?")

    if seq >= current.sequence:
        LOG.info("Database already up-to-date.")
        return 0

    remote_server_age_sec = int((dt.datetime.now(dt.timezone.utc) - current.timestamp).total_seconds())
    LOG.debug("Applying %d sequence(s) (%d → %d), covering %s (%s sec) of changes (%s → %s)",
        current.sequence - seq, current.sequence, seq,
        pretty_format_timedelta((current.timestamp - ts).total_seconds()),
        int((current.timestamp - ts).total_seconds()),
        osm_date(ts.astimezone(dt.timezone.utc)),
        osm_date(current.timestamp.astimezone(dt.timezone.utc))
    )

    update_started = dt.datetime.now(dt.timezone.utc)

    if args.diff_file is not None:
        outfile = Path(args.diff_file)
    else:
        tmpdir = tempfile.TemporaryDirectory()
        outfile = Path(tmpdir.name) / 'osm2pgsql_diff.osc.gz'

    osm2pgsql = [args.osm2pgsql_cmd, '--append', '--slim', '--prefix', args.prefix]
    osm2pgsql.extend(args.extra_params)
    if args.middle_schema != 'public':
        osm2pgsql.extend(('--middle-schema', args.middle_schema))
    if args.database:
        osm2pgsql.extend(('-d', args.database))
    if args.username:
        osm2pgsql.extend(('-U', args.username))
    if args.host:
        osm2pgsql.extend(('-H', args.host))
    if args.port:
        osm2pgsql.extend(('-P', args.port))
    osm2pgsql.append(str(outfile))
    LOG.debug("Calling osm2pgsql with: %s", ' '.join(osm2pgsql))

    while seq < current.sequence:
        LOG.debug("Importing from sequence %d", seq)
        if outfile.exists():
            outfile.unlink()
        outhandler = WriteHandler(str(outfile))
        endseq = repl.apply_diffs(outhandler, seq + 1,
                                  max_size=args.max_diff_size * 1024)
        outhandler.close()

        if endseq is None:
            LOG.debug("No new diffs found.")
            break

        subprocess.run(osm2pgsql, check=True)
        seq = endseq

        nextstate = repl.get_state_info(seq)
        timestamp = nextstate.timestamp if nextstate else None

        if args.post_processing:
            cmd = [args.post_processing, str(endseq), str(timestamp or '')]
            LOG.debug('Calling post-processing script: %s', ' '.join(cmd))
            subprocess.run(cmd, check=True)

        props.write_replication_state(base_url, seq, nextstate.timestamp if nextstate else None)

        if nextstate is not None:
            LOG.info("Data imported until %s. Backlog remaining: %s",
                osm_date(nextstate.timestamp.astimezone(dt.timezone.utc)),
                pretty_format_timedelta((dt.datetime.now(dt.timezone.utc) - nextstate.timestamp).total_seconds()),
            )

        if args.once:
            break

    update_duration_sec = (dt.datetime.now(dt.timezone.utc) - update_started).total_seconds()
    _base_url, _seq, current_local_timestamp = props.get_replication_state()

    total_applied_changes_duration_sec = (current_local_timestamp - initial_local_timestamp).total_seconds()
    LOG.debug("It took %s (%d sec) to apply %s (%d sec) of changes. This is a speed of ×%.1f.",
            pretty_format_timedelta(update_duration_sec), int(update_duration_sec),
            pretty_format_timedelta(total_applied_changes_duration_sec), int(total_applied_changes_duration_sec),
            total_applied_changes_duration_sec / update_duration_sec
        )

    return 0


def get_parser():
    parser = ArgumentParser(description=__doc__,
                            prog='osm2pgsql-replication',
                            formatter_class=RawDescriptionHelpFormatter)
    subs = parser.add_subparsers(title='available commands', dest='subcommand')

    default_args = ArgumentParser(add_help=False)
    group = default_args.add_argument_group('Default arguments')
    group.add_argument('-h', '--help', action='help',
                       help='Show this help message and exit')
    group.add_argument('-q', '--quiet', action='store_const', const=0,
                       dest='verbose', default=2,
                       help='Print only error messages')
    group.add_argument('-v', '--verbose', action='count', default=2,
                       help='Increase verboseness of output')
    group = default_args.add_argument_group('Database arguments',
      "The following arguments can be used to set the connection parameters to the\n"
      "osm2pgsql database. You may also use libpq environment variables to set\n"
      "connection parameters, see https://www.postgresql.org/docs/current/libpq-envars.html.\n"
      "If your database connection requires a password, use a pgpass file,\n"
      "see https://www.postgresql.org/docs/current/libpq-pgpass.html.")
    group.add_argument('-d', '--database', metavar='DB',
                       help='Name of PostgreSQL database to connect to or conninfo string')
    group.add_argument('-U', '--username', metavar='NAME',
                       help='PostgreSQL user name')
    group.add_argument('-H', '--host', metavar='HOST',
                       help='Database server host name or socket location')
    group.add_argument('-P', '--port', metavar='PORT',
                       help='Database server port')
    group.add_argument('-p', '--prefix', metavar='PREFIX', default='planet_osm',
                       help="Prefix for table names (default 'planet_osm')")
    group.add_argument('--middle-schema', metavar='SCHEMA', default=None,
                    help='Name of the schema to store the table for the replication state in')
    group.add_argument('--schema', metavar='SCHEMA', default=None,
                    help='Name of the schema for the database')

    # Arguments for init
    cmd = subs.add_parser('init', parents=[default_args],
                          help=init.__doc__.split('\n', 1)[0],
                          description=dedent(init.__doc__),
                          formatter_class=RawDescriptionHelpFormatter,
                          add_help=False)
    grp = cmd.add_argument_group('Replication source arguments')
    srcgrp = grp.add_mutually_exclusive_group()
    srcgrp.add_argument('--osm-file', metavar='FILE',
                        help='Get replication information from the given file.')
    srcgrp.add_argument('--server', metavar='URL',
                        help='Use replication server at the given URL')
    grp.add_argument('--start-at', metavar='TIME', type=start_point,
                     help='Time when to start replication. When an absolute timestamp '
                          '(in ISO format) is given, it will be used. If a number '
                          'is given, then replication starts the number of minutes '
                          'before the known date of the database.')

    cmd.set_defaults(handler=init)

    # Arguments for update
    cmd = subs.add_parser('update', parents=[default_args],
                          usage='%(prog)s update [options] [-- param [param ...]]',
                          help=update.__doc__.split('\n', 1)[0],
                          description=dedent(update.__doc__),
                          formatter_class=RawDescriptionHelpFormatter,
                          add_help=False)
    cmd.set_defaults(handler=update)
    cmd.add_argument('extra_params', nargs='*', metavar='param',
                     help='Extra parameters to hand in to osm2pgsql.')
    grp = cmd.add_argument_group('Update process arguments')
    cmd.add_argument('--diff-file', metavar='FILE',
                     help='File to save changes before they are applied to osm2pgsql.')
    cmd.add_argument('--max-diff-size', type=int, default=500,
                     help='Maximum data to load in MB (default: 500MB)')
    cmd.add_argument('--osm2pgsql-cmd', default=str(OSM2PGSQL_PATH),
                     help=f'Path to osm2pgsql command (default: {OSM2PGSQL_PATH})')
    cmd.add_argument('--once', action='store_true',
                     help='Run updates only once, even when more data is available.')
    cmd.add_argument('--post-processing', metavar='SCRIPT',
                     help='Post-processing script to run after each execution of osm2pgsql.')

    # Arguments for status
    cmd = subs.add_parser('status', parents=[default_args],
                          help=status.__doc__.split('\n', 1)[0],
                          description=dedent(status.__doc__),
                          formatter_class=RawDescriptionHelpFormatter,
                          add_help=False)
    cmd.add_argument('--json', action="store_true", default=False, help="Output status as json.")
    cmd.set_defaults(handler=status)

    return parser


def main(prog_args=None):
    parser = get_parser()
    try:
        args = parser.parse_args(args=prog_args)
    except SystemExit:
        return 1

    if missing_modules:
        LOG.fatal("Missing required Python libraries %(mods)s.\n\n"
                  "To install them via pip run: pip install %(mods)s\n\n"
                  "ERROR: libraries could not be loaded.",
                  dict(mods=" ".join(missing_modules)))
        return 1


    if args.subcommand is None:
        parser.print_help()
        return 1

    logging.basicConfig(stream=sys.stderr,
                        format='{asctime} [{levelname}]: {message}',
                        style='{',
                        datefmt='%Y-%m-%d %H:%M:%S',
                        level=max(4 - args.verbose, 1) * 10)

    args.middle_schema = args.middle_schema or args.schema or 'public'

    with DBConnection(args) as db:
        if db.table_exists(Osm2pgsqlProperties.PROP_TABLE_NAME):
            props = Osm2pgsqlProperties(db)
        else:
            props = LegacyProperties(db, args.prefix)

        if not props.is_updatable:
            LOG.fatal(f'osm2pgsql middle table "{args.middle_schema}.{args.prefix}_ways" not found in database "{db.name}". '
                       'Database needs to be imported in --slim mode.')
            return 1

        try:
            return args.handler(props, args)
        except DBError as err:
            LOG.fatal(err.msg)

    return 1


if __name__ == '__main__':
    try:
        retcode = main()
    except Exception as ex:
        LOG.fatal("Exception during execution: %s", ex)
        retcode = 3

    sys.exit(retcode)