nominatim/api.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2022 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Implementation of classes for API access via libraries.
   9 """
  10 from typing import Mapping, Optional
  11 import asyncio
  12 from pathlib import Path
  13
  14 from sqlalchemy.engine.url import URL
  15 from sqlalchemy.ext.asyncio import create_async_engine
  16
  17 from nominatim.config import Configuration
  18 from nominatim.apicmd.status import get_status, StatusResult
  19
  20 class NominatimAPIAsync:
  21     """ API loader asynchornous version.
  22     """
  23     def __init__(self, project_dir: Path,
  24                  environ: Optional[Mapping[str, str]] = None) -> None:
  25         self.config = Configuration(project_dir, environ)
  26
  27         dsn = self.config.get_database_params()
  28
  29         dburl = URL.create(
  30                    'postgresql+asyncpg',
  31                    database=dsn.get('dbname'),
  32                    username=dsn.get('user'), password=dsn.get('password'),
  33                    host=dsn.get('host'), port=int(dsn['port']) if 'port' in dsn else None,
  34                    query={k: v for k, v in dsn.items()
  35                           if k not in ('user', 'password', 'dbname', 'host', 'port')})
  36         self.engine = create_async_engine(dburl,
  37                                           connect_args={"server_settings": {"jit": "off"}},
  38                                           future=True)
  39
  40
  41     async def status(self) -> StatusResult:
  42         """ Return the status of the database.
  43         """
  44         return await get_status(self.engine)
  45
  46
  47 class NominatimAPI:
  48     """ API loader, synchronous version.
  49     """
  50
  51     def __init__(self, project_dir: Path,
  52                  environ: Optional[Mapping[str, str]] = None) -> None:
  53         self.async_api = NominatimAPIAsync(project_dir, environ)
  54
  55
  56     def status(self) -> StatusResult:
  57         """ Return the status of the database.
  58         """
  59         return asyncio.get_event_loop().run_until_complete(self.async_api.status())