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, TypeVar, Callable, Any
  11 import functools
  12 import asyncio
  13 from pathlib import Path
  14
  15 from sqlalchemy.engine.url import URL
  16 from sqlalchemy.ext.asyncio import create_async_engine
  17
  18 from nominatim.typing import StrPath
  19 from nominatim.config import Configuration
  20 from nominatim.apicmd.status import get_status, StatusResult
  21
  22 class NominatimAPIAsync:
  23     """ API loader asynchornous version.
  24     """
  25     def __init__(self, project_dir: Path,
  26                  environ: Optional[Mapping[str, str]] = None) -> None:
  27         self.config = Configuration(project_dir, environ)
  28
  29         dsn = self.config.get_database_params()
  30
  31         dburl = URL.create(
  32                    'postgresql+asyncpg',
  33                    database=dsn.get('dbname'),
  34                    username=dsn.get('user'), password=dsn.get('password'),
  35                    host=dsn.get('host'), port=int(dsn['port']) if 'port' in dsn else None,
  36                    query={k: v for k, v in dsn.items()
  37                           if k not in ('user', 'password', 'dbname', 'host', 'port')})
  38         self.engine = create_async_engine(dburl,
  39                                           connect_args={"server_settings": {"jit": "off"}},
  40                                           future=True)
  41
  42
  43     async def status(self) -> StatusResult:
  44         """ Return the status of the database.
  45         """
  46         return await get_status(self.engine)
  47
  48
  49 class NominatimAPI:
  50     """ API loader, synchronous version.
  51     """
  52
  53     def __init__(self, project_dir: Path,
  54                  environ: Optional[Mapping[str, str]] = None) -> None:
  55         self.async_api = NominatimAPIAsync(project_dir, environ)
  56
  57
  58     def status(self) -> StatusResult:
  59         return asyncio.get_event_loop().run_until_complete(self.async_api.status())