python类_solve_conn_curs()的实例源码

def _from_db(self, name, pyrange, conn_or_curs):
        """Return a `RangeCaster` instance for the type *pgrange*.

        Raise `ProgrammingError` if the type is not found.
        """
        from psycopg2.extensions import STATUS_IN_TRANSACTION
        from psycopg2.extras import _solve_conn_curs
        conn, curs = _solve_conn_curs(conn_or_curs)

        if conn.server_version < 90200:
            raise ProgrammingError("range types not available in version %s"
                % conn.server_version)

        # Store the transaction status of the connection to revert it after use
        conn_status = conn.status

        # Use the correct schema
        if '.' in name:
            schema, tname = name.split('.', 1)
        else:
            tname = name
            schema = 'public'

        # get the type oid and attributes
        try:
            curs.execute("""\
select rngtypid, rngsubtype,
    (select typarray from pg_type where oid = rngtypid)
from pg_range r
join pg_type t on t.oid = rngtypid
join pg_namespace ns on ns.oid = typnamespace
where typname = %s and ns.nspname = %s;
""", (tname, schema))

        except ProgrammingError:
            if not conn.autocommit:
                conn.rollback()
            raise
        else:
            rec = curs.fetchone()

            # revert the status of the connection as before the command
            if (conn_status != STATUS_IN_TRANSACTION
            and not conn.autocommit):
                conn.rollback()

        if not rec:
            raise ProgrammingError(
                "PostgreSQL type '%s' not found" % name)

        type, subtype, array = rec

        return RangeCaster(name, pyrange,
            oid=type, subtype_oid=subtype, array_oid=array)