代码之家 › 专栏 › 技术社区 › Tiago Duque

如何使用Postgres在SQLAlchemy中将值转换为自定义类型

sqlalchemy python

Tiago Duque · 技术社区 · 4 年前

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

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进行迁移时,它停止了工作。但是,数据库结构完全相同。

0 回复 | 直到 4 年前

Danila Ganchar 4 年前

我曾经遇到过同样的问题。首先,我可以建议更新 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 .