pg.py 文件源码

def __init__(self, db):
        """Initialize type cache for connection."""
        super(DbTypes, self).__init__()
        self._db = weakref.proxy(db)
        self._regtypes = False
        self._typecasts = Typecasts()
        self._typecasts.get_attnames = self.get_attnames
        self._typecasts.connection = self._db
        if db.server_version < 80400:
            # older remote databases (not officially supported)
            self._query_pg_type = (
                "SELECT oid, typname, typname::text::regtype,"
                " typtype, null as typcategory, typdelim, typrelid"
                " FROM pg_type WHERE oid=%s::regtype")
        else:
            self._query_pg_type = (
                "SELECT oid, typname, typname::regtype,"
                " typtype, typcategory, typdelim, typrelid"
                " FROM pg_type WHERE oid=%s::regtype")