Changeset 788


Ignore:
Timestamp:
Jan 26, 2016, 4:13:55 PM (4 years ago)
Author:
cito
Message:

Add method columns() to type cache

The method returns the columns of composite types.
This makes the type cache even more useful.

Location:
trunk
Files:
3 edited

Legend:

Unmodified
Added
Removed
  • trunk/docs/contents/pgdb/connection.rst

    r787 r788  
    9999<http://www.postgresql.org/docs/current/static/catalog-pg-type.html>`_.
    100100
     101The :attr:`Connection.type_cache` also provides a method :meth:`columns`
     102that returns the names and type OIDs of the columns of composite types.
     103
    101104.. versionadded:: 5.0
  • trunk/pgdb.py

    r787 r788  
    160160    ['oid', 'name', 'len', 'type', 'category', 'delim', 'relid'])
    161161
     162ColumnInfo = namedtuple('ColumnInfo', ['name', 'type'])
     163
    162164
    163165class TypeCache(dict):
     
    175177
    176178    def __missing__(self, key):
    177         q = ("SELECT oid, typname,"
     179        """Get the type info from the database if it is not cached."""
     180        if isinstance(key, int):
     181            oid = key
     182        else:
     183            if not '.' in key and not '"' in key:
     184                key = '"%s"' % key
     185            oid = "'%s'::regtype" % self._escape_string(key)
     186        self._src.execute("SELECT oid, typname,"
    178187             " typlen, typtype, typcategory, typdelim, typrelid"
    179             " FROM pg_type WHERE ")
    180         if isinstance(key, int):
    181             q += "oid = %d" % key
    182         else:
    183             q += "typname = '%s'" % self._escape_string(key)
    184         self._src.execute(q)
     188            " FROM pg_type WHERE oid=%s" % oid)
    185189        res = self._src.fetch(1)
    186190        if not res:
     
    193197        self[res.oid] = self[res.name] = res
    194198        return res
     199
     200    def columns(self, key):
     201        """Get the names and types of the columns of composite types."""
     202        typ = self[key]
     203        if typ.type != 'c' or not typ.relid:
     204            return []  # this type is not composite
     205        self._src.execute("SELECT attname, atttypid"
     206            " FROM pg_attribute WHERE attrelid=%s AND attnum>0"
     207            " AND NOT attisdropped ORDER BY attnum" % typ.relid)
     208        return [ColumnInfo(name, int(oid))
     209            for name, oid in self._src.fetch(-1)]
    195210
    196211    @staticmethod
  • trunk/tests/test_dbapi20.py

    r787 r788  
    294294        con = self._connect()
    295295        cur = con.cursor()
    296         type_info = cur.type_cache['numeric']
     296        type_cache = cur.type_cache
     297        type_info = type_cache['numeric']
    297298        self.assertEqual(type_info.oid, 1700)
    298299        self.assertEqual(type_info.name, 'numeric')
     
    301302        self.assertEqual(type_info.delim, ',')
    302303        self.assertIs(cur.type_cache[1700], type_info)
     304        type_info = type_cache['pg_type']
     305        self.assertEqual(type_info.type, 'c')  # composite
     306        self.assertEqual(type_info.category, 'C')  # composite
     307        cols = type_cache.columns('pg_type')
     308        self.assertEqual(cols[0].name, 'typname')
     309        typname = type_cache[cols[0].type]
     310        self.assertEqual(typname.name, 'name')
     311        self.assertEqual(typname.type, 'b')  # base
     312        self.assertEqual(typname.category, 'S')  # string
     313        self.assertEqual(cols[3].name, 'typlen')
     314        typlen = type_cache[cols[3].type]
     315        self.assertEqual(typlen.name, 'int2')
     316        self.assertEqual(typlen.type, 'b')  # base
     317        self.assertEqual(typlen.category, 'N')  # numeric
    303318
    304319    def test_cursor_iteration(self):
Note: See TracChangeset for help on using the changeset viewer.