codekingpro/portable-devtools

1"""2psycopg libpq wrapper3 4This package exposes the libpq functionalities as Python objects and functions.5 6The real implementation (the binding to the C library) is7implementation-dependant but all the implementations share the same interface.8"""9 10# Copyright (C) 2020 The Psycopg Team11 12import os13import logging14from typing import Callable, List, Type15 16from . import abc17from .misc import ConninfoOption, PGnotify, PGresAttDesc18from .misc import error_message19from ._enums import ConnStatus, DiagnosticField, ExecStatus, Format, Trace20from ._enums import Ping, PipelineStatus, PollingStatus, TransactionStatus21 22logger = logging.getLogger(__name__)23 24__impl__: str25"""The currently loaded implementation of the `!psycopg.pq` package.26 27Possible values include ``python``, ``c``, ``binary``.28"""29 30__build_version__: int31"""The libpq version the C package was built with.32 33A number in the same format of `~psycopg.ConnectionInfo.server_version`34representing the libpq used to build the speedup module (``c``, ``binary``) if35available.36 37Certain features might not be available if the built version is too old.38"""39 40version: Callable[[], int]41PGconn: Type[abc.PGconn]42PGresult: Type[abc.PGresult]43Conninfo: Type[abc.Conninfo]44Escaping: Type[abc.Escaping]45PGcancel: Type[abc.PGcancel]46 47 48def import_from_libpq() -> None:49    """50    Import pq objects implementation from the best libpq wrapper available.51 52    If an implementation is requested try to import only it, otherwise53    try to import the best implementation available.54    """55    # import these names into the module on success as side effect56    global __impl__, version, __build_version__57    global PGconn, PGresult, Conninfo, Escaping, PGcancel58 59    impl = os.environ.get("PSYCOPG_IMPL", "").lower()60    module = None61    attempts: List[str] = []62 63    def handle_error(name: str, e: Exception) -> None:64        if not impl:65            msg = f"couldn't import psycopg '{name}' implementation: {e}"66            logger.debug(msg)67            attempts.append(msg)68        else:69            msg = f"couldn't import requested psycopg '{name}' implementation: {e}"70            raise ImportError(msg) from e71 72    # The best implementation: fast but requires the system libpq installed73    if not impl or impl == "c":74        try:75            from psycopg_c import pq as module  # type: ignore76        except Exception as e:77            handle_error("c", e)78 79    # Second best implementation: fast and stand-alone80    if not module and (not impl or impl == "binary"):81        try:82            from psycopg_binary import pq as module  # type: ignore83        except Exception as e:84            handle_error("binary", e)85 86    # Pure Python implementation, slow and requires the system libpq installed.87    if not module and (not impl or impl == "python"):88        try:89            from . import pq_ctypes as module  # type: ignore[assignment]90        except Exception as e:91            handle_error("python", e)92 93    if module:94        __impl__ = module.__impl__95        version = module.version96        PGconn = module.PGconn97        PGresult = module.PGresult98        Conninfo = module.Conninfo99        Escaping = module.Escaping100        PGcancel = module.PGcancel101        __build_version__ = module.__build_version__102    elif impl:103        raise ImportError(f"requested psycopg implementation '{impl}' unknown")104    else:105        sattempts = "\n".join(f"- {attempt}" for attempt in attempts)106        raise ImportError(107            f"""\108no pq wrapper available.109Attempts made:110{sattempts}"""111        )112 113 114import_from_libpq()115 116__all__ = (117    "ConnStatus",118    "PipelineStatus",119    "PollingStatus",120    "TransactionStatus",121    "ExecStatus",122    "Ping",123    "DiagnosticField",124    "Format",125    "Trace",126    "PGconn",127    "PGnotify",128    "Conninfo",129    "PGresAttDesc",130    "error_message",131    "ConninfoOption",132    "version",133)134