test/bdd/utils/db.py

   1 # SPDX-License-Identifier: GPL-3.0-or-later
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2025 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Helper functions for managing test databases.
   9 """
  10 import asyncio
  11 import psycopg
  12 from psycopg import sql as pysql
  13
  14 from nominatim_db.tools.database_import import setup_database_skeleton, create_tables, \
  15                                                create_partition_tables, create_search_indices
  16 from nominatim_db.data.country_info import setup_country_tables
  17 from nominatim_db.tools.refresh import create_functions, load_address_levels_from_config
  18 from nominatim_db.tools.exec_utils import run_osm2pgsql
  19 from nominatim_db.tokenizer import factory as tokenizer_factory
  20
  21 class DBManager:
  22
  23     def __init__(self, purge=False):
  24         self.purge = purge
  25
  26     def check_for_db(self, dbname):
  27         """ Check if the given DB already exists.
  28             When the purge option is set, then an existing database will
  29             be deleted and the function returns that it does not exist.
  30         """
  31         if self.purge:
  32             self.drop_db(dbname)
  33             return False
  34
  35         return self.exists_db(dbname)
  36
  37     def drop_db(self, dbname):
  38         """ Drop the given database if it exists.
  39         """
  40         with psycopg.connect(dbname='postgres') as conn:
  41             conn.autocommit = True
  42             conn.execute(pysql.SQL('DROP DATABASE IF EXISTS')
  43                          + pysql.Identifier(dbname))
  44
  45     def exists_db(self, dbname):
  46         """ Check if a database with the given name exists already.
  47         """
  48         with psycopg.connect(dbname='postgres') as conn:
  49             cur = conn.execute('select count(*) from pg_database where datname = %s',
  50                                (dbname,))
  51             return cur.fetchone()[0] == 1
  52
  53     def create_db_from_template(self, dbname, template):
  54         """ Create a new database from the given template database.
  55             Any existing database with the same name will be dropped.
  56         """
  57         with psycopg.connect(dbname='postgres') as conn:
  58             conn.autocommit = True
  59             conn.execute(pysql.SQL('DROP DATABASE IF EXISTS')
  60                          + pysql.Identifier(dbname))
  61             conn.execute(pysql.SQL('CREATE DATABASE {} WITH TEMPLATE {}')
  62                               .format(pysql.Identifier(dbname),
  63                                       pysql.Identifier(template)))
  64
  65     def setup_template_db(self, config):
  66         """ Create a template DB which contains the necessary extensions
  67             and basic static tables.
  68
  69             The template will only be created if the database does not yet
  70             exist or 'purge' is set.
  71         """
  72         dsn = config.get_libpq_dsn()
  73
  74         if self.check_for_db(config.get_database_params()['dbname']):
  75             return
  76
  77         setup_database_skeleton(dsn)
  78
  79         run_osm2pgsql(dict(osm2pgsql='osm2pgsql',
  80                            osm2pgsql_cache=1,
  81                            osm2pgsql_style=str(config.get_import_style_file()),
  82                            osm2pgsql_style_path=config.lib_dir.lua,
  83                            threads=1,
  84                            dsn=dsn,
  85                            flatnode_file='',
  86                            tablespaces=dict(slim_data='', slim_index='',
  87                                             main_data='', main_index=''),
  88                            append=False,
  89                            import_data=b'<osm version="0.6"></osm>'))
  90
  91         setup_country_tables(dsn, config.lib_dir.data)
  92
  93         with psycopg.connect(dsn) as conn:
  94             create_tables(conn, config)
  95             load_address_levels_from_config(conn, config)
  96             create_partition_tables(conn, config)
  97             create_functions(conn, config, enable_diff_updates=False)
  98             asyncio.run(create_search_indices(conn, config))
  99
 100         tokenizer_factory.create_tokenizer(config)
 101