使用flask-sqlalchemy，如何在数据库表列中指定用户定义的数据类型答案

【问题标题】：Using flask-sqlalchemy, how to specify user-defined data type in database table column使用flask-sqlalchemy，如何在数据库表列中指定用户定义的数据类型
【发布时间】：2021-10-31 06:20:31
【问题描述】：

我正在用 python 编写一个与 Postgres 数据库对话的 rest api 服务。我正在使用 flake-sqlalchemy 这样做。在表中，其中一列定义为：

my_column my_type not null

并且定义了 my_type：

create type my_type as emum ('val1', 'val2', 'val3')

如何在 flask-sqlalchemy 中表示这个？谢谢你的帮助。

【问题讨论】：

标签： python postgresql rest flask flask-sqlalchemy

【解决方案1】：

您的 app.py 定义如下：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from sqlalchemy.types import TypeDecorator, CHAR

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://test:test@localhost/test?charset=utf8mb4' 
db = SQLAlchemy(app)
Migrate(app, db, compare_type=True)

class CustEnum(TypeDecorator):
    impl = CHAR(4)
    def process_bind_param(self, value, dialect):
        valid_values = ('val1', 'val2', 'val3')
        if value not in valid_values:
            raise TypeError(f"mismatch {value} not in {valid_values} ")
            
        return value

class Blobby(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    type_col = db.Column(CustEnum())

process_bind_param 方法基本上是 setter 方法，process_result_value 是 getter 方法。

我还想用 cli 装饰器向文件中添加一个测试函数来检查它是否有效。

@app.cli.command("testit")
def testit():
    newblob = Blobby(type_col = 'val2')
    db.session.add( newblob )
    db.session.commit()
    
    for row in db.session.execute('select * from blobby'):
        print(row)
    
    input('press enter to continue to failure test')
    # value not allowed should fail
    newblob = Blobby(type_col = 'fail')
    db.session.add( newblob )
    db.session.commit()

当您应用迁移（即运行 flask db init、flask db migrate 和 flask db upgrade）时，您会收到一个错误，即找不到名称 app.CustEnum。

sa.Column('type_col', app.CustEnum(length=4), nullable=True),
NameError: name 'app' is not defined

在寻找解决方案时，我发现库的作者说您需要更新迁移脚本本身，这有点落后于 IMO。

https://github.com/miguelgrinberg/Flask-Migrate/issues/282#issuecomment-518989508

所以只需编辑migrations/versions/98f29efa609f_.py 或任何您的版本。并添加 import app 或任何你的烧瓶脚本的命名空间。

最后一件事是用flask testit 运行testit() 函数给你一个成功和一个失败。

sqlalchemy.exc.StatementError: (builtins.TypeError) mismatch fail not in ('val1', 'val2', 'val3')
[SQL: INSERT INTO blobby (type_col) VALUES (%(type_col)s)]
[parameters: [{'type_col': 'fail'}]]

【讨论】：