如何使用 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、Flask-SQLAlchemy==2.4.0、SQLAlchemy==1.3.18、SQLAlchemy-Utils==0.36.6 工作正常：

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。