Function _connect

psycopg/psycopg/generators.py:62–110

Generator to create a database connection without blocking.

(conninfo: str, *, timeout: float = 0.0)

def _connect(conninfo: str, *, timeout: float = 0.0) -> PQGenConn[PGconn]:
    """
    Generator to create a database connection without blocking.
    """
    deadline = monotonic() + timeout if timeout else 0.0

    # To debug slowdown during connection:
    #
    #   $ PSYCOPG_IMPL=python python
    #   >>> import logging
    #   >>> logging.basicConfig(level=logging.DEBUG, format='%(asctime)s %(message)s')
    #   >>> logging.getLogger("psycopg").setLevel(logging.DEBUG)

    conn = pq.PGconn.connect_start(conninfo.encode())
    logger.debug("connection started: %s", conn)
    while True:
        if conn.status == BAD:
            encoding = conninfo_encoding(conninfo)
            raise e.OperationalError(
                f"connection is bad: {conn.get_error_message(encoding)}", pgconn=conn
            )

        status = conn.connect_poll()
        logger.debug("connection polled: %s", conn)

        if status == POLL_READING or status == POLL_WRITING:
            wait = WAIT_R if status == POLL_READING else WAIT_W
            while True:
                ready = yield conn.socket, wait
                if deadline and monotonic() > deadline:
                    raise e.ConnectionTimeout("connection timeout expired")
                if ready:
                    break

        elif status == POLL_OK:
            break
        elif status == POLL_FAILED:
            encoding = conninfo_encoding(conninfo)
            raise e.OperationalError(
                f"connection failed: {conn.get_error_message(encoding)}",
                pgconn=e.finish_pgconn(conn),
            )
        else:
            raise e.InternalError(
                f"unexpected poll status: {status}", pgconn=e.finish_pgconn(conn)
            )

    conn.nonblocking = 1
    return conn


def _cancel(cancel_conn: PGcancelConn, *, timeout: float = 0.0) -> PQGenConn[None]: