如何在带Postgres的SQLAlchemy中将值强制转换为自定义类型

Question

在我正在构建的应用程序中，我使用了自定义类型的组合类型来存储数据。 它基本上是一个简单类型对象的列表：

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

这在我的课堂上是以如下方式建立的：

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())])))

然后我尝试：

   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

突然，我发现了这个错误：

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.

但是我找不到怎么做这个演员

我应该创建一个类来描述postgres中使用的这种类型，然后在列表中实例化它吗？如果是，请举例说明

顺便说一句，这曾经奏效。使用flask db migrate进行迁移时停止工作。但是，数据库结构完全相同

Answer 1

我曾经有过同样的问题。首先，我可以建议更新Flask、Flask-SQLAlchemy和相关软件包Flask==1.1.2，{}，{}，{}工作正常：

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'

但是几年前，我用sqlalchemy_utils.register_composites()解决了这个问题。是这样的：

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

你也可以试试psycopg2.extras.register_composite