How to cast values to custom type in SQLAlchemy with Postgres

Question

In the application that I'm building, I've used a Composite Type of Custom types to store data. It's basically a list of simple objects of type:

bin = CompositeType(columns=[Column('label', String()), Column('head', Float()), Column('tail', Float())], name='bin')

This is built into my class in a way such as:

class Myclass(db.Model):
   id = sa.Column('id', sa.Integer(), nullable=False)
   bin = sa.Column('bin', CompositeArray(CompositeType('bin', [Column('label', String()), Column('head', Float()), Column('tail', Float())])))

Then I try:

   id = '111331'
   bins ={'1':{'start':0,'end':1},'2':{'start':1,'end':2}}
   myclass = Myclass(id = int(id), bins = [(bin_key, bins[bin_key]['start'], bins[bin_key]['end']) for bin_key in bins])
   # The object creation works flawlessly.
   db.session.add(myclass)
   db.session.commit()
   # Error happens in SQLAlchemy commit

Suddenly, I'm getting this error:

sqlalchemy.exc.ProgrammingError: (psycopg2.errors.DatatypeMismatch)
column "bin" is of type bin[] but expression is of type record[] (...)
^ HINT:  You will need to rewrite or cast the expression.

But I can't manage to find how to do this casting.

Should I make a class to describe this type used in postgres and then instantiate it in the list comprehension? If so, please, give examples.

Btw, this used to work. Stopped working when I migrated using flask db migrate. However, database structure is exactly the same.

Answer 1

I had the same problem once. First of all I can recommend to update Flask, Flask-SQLAlchemy and related packages. Flask==1.1.2, Flask-SQLAlchemy==2.4.0, SQLAlchemy==1.3.18, SQLAlchemy-Utils==0.36.6 works fine:

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgres://postgres:postgres@localhost:5432/test'
db = SQLAlchemy(app)


class Myclass(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    custom_type = db.Column(
       'custom_type',
       CompositeArray(
           CompositeType(
                'custom_type',
                [
                    db.Column('id', db.Text),
                    db.Column('name', db.Text),
                ]
           )
       )
    )


db.create_all()
db.engine.execute("""
    DO $$
    BEGIN
    IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'custom_type') THEN
        CREATE TYPE custom_type AS (
            id text,
            name text
        );
    END IF;
    END$$;
""")


@app.route('/')
def test():
    db.session.add(Myclass(custom_type=[
        {'id': 'test_1', 'name': 'first'},
        {'id': 'test_2', 'name': 'second'},
    ]))

    db.session.commit()
    return 'done'

But a few years ago I solved the problem using sqlalchemy_utils.register_composites(). It was something like this:

def create_app():
    app = Flask(__name__)
    # blablabla
    engine = db.get_engine()
    engine.dispose()
    register_composites(engine.connect())
    return app

Also you can try psycopg2.extras.register_composite.