what is the psycopg 3.0.9 equivalent of composite casters in psycopg2?

Question

   import psycopg
   from psycopg.types.composite import CompositeInfo, register_composite

   def insert_data(self):
      pg = psycopg.connect(self.connection_string)
      cursor = pg.cursor()
      
      class TradeComposite(CompositeInfo):
        def __init__(self, name, oid, attrs, array_oid=None, schema=None):
          self.name = name
          self.schema = schema
          self.oid = oid
          self.array_oid = array_oid

          self.attnames = [a[0] for a in attrs]
          self.atttypes = [a[1] for a in attrs]
          # self._create_type(name, self.attnames)
          # self.typecaster = _ext.new_type((oid,), name, self.parse)
          # if array_oid:
          #     self.array_typecaster = _ext.new_array_type(
          #         (array_oid,), "%sARRAY" % name, self.typecaster)
          # else:
          #     self.array_typecaster = None
        def make(self, values):
          return tuple(values)
      
      info = CompositeInfo.fetch(pg, 'insertable_unique_trade')
      register_composite(info, cursor, factory=TradeComposite)

      def handle_messages():
          trades = [
          (55, 'acctname', 0, 69, 66376.07596266, 68, 66386.70210866087, 68, 0.0, 1.00016009, '6043080995', 1641444405.716349), 
          (55, 'acctname', 0, 69, 2.05263549, 68, 6750.199985115853, 68, 0.0, 3288.55270115, '6215459421', 1642183827.224316)
          ]
          cursor.execute("SELECT insert_unique_trades(%s::insertable_unique_trade[])::insertable_unique_trade[]", [trades])
          results = cursor.fetchall()
          pg.commit()

the composite type was created in the psql script with CREATE TYPE insertable_unique_trade AS

I can insert the data to the db with pg.commit() but i cant perform cursor.fetchall() because of the (I think) TradeComposite class. Whats the equivalent implementation of psycopg2 composite casters in psycopg 3.0.9? i had to upgrade from psycopg2 2.9.3 to psycopg 3.0.9.

Answer 1

You can simply use this:

type_info = CompositeInfo.fetch(conn=self.Connection, name="your_type")
register_composite(info=type_info, context=self.Connection, factory=None)

And you will get the result type as a namedtuple.