小编Clo*_*loC的帖子

我想使用 alembic 运行从 sqlalchemy 模型到另一个模型的迁移。初始模型如下所示：

from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy_utils import UUIDType

class Foo(db.Model):
    __tablename__ = 'foo'
    id = db.Column(UUIDType(binary=True), primary_key=True)
    foo_field = db.Column(JSONB)

我想让它foo_field不可为空：

class Foo(db.Model):
    __tablename__ = 'foo'
    id = db.Column(UUIDType(binary=True), primary_key=True)
    foo_field = db.Column(JSONB, nullable=False)

我想运行 alembic 脚本来更改列，但由于我现有的一些列为foo_field空，所以我想server_default在更改时应用默认值（空字典）

op.alter_column('foo',
         'foo_field',
         existing_type=postgresql.JSONB(astext_type=sa.Text()),
         nullable=False,
         server_default="{}")

我尝试了不同的选项来传递给这个server_default，例如text("{}")，，，，。但它似乎被忽略了，我得到的只是 运行升级时的：lambda: {}{}text("SELECT '{}'")IntegrityError

sqlalchemy.exc.IntegrityError: (psycopg2.IntegrityError) column "foo_field" contains null values [SQL: 'ALTER TABLE foo ALTER COLUMN foo_field SET …

嗨，下面是我的运行代码，可以通过以下 URL 访问：http : //127.0.0.1 : 5000/api/documentation

from flask import Flask, Blueprint
from flask_restplus import Api, Resource, fields

app = Flask(__name__)
blueprint = Blueprint('api', __name__, url_prefix='/api')
api = Api(blueprint, doc='/documentation') #,doc=False

app.register_blueprint(blueprint)

app.config['SWAGGER_UI_JSONEDITOR'] = True

login_details = api.model('LoginModel',{'user_name' : fields.String('The Username.'),'pass_word' : fields.String('The password.'),})
# pass_word = api.model('Pwd', {'pass_word' : fields.String('The password.')})
credentials = []
python = {'user_name' : '1234','pwd':'23213413'}
credentials.append(python)

@api.route('/login')
class Language(Resource):

    @api.marshal_with(login_details, envelope='the_data',mask='pass_word')
    def get(self):
        return credentials

    @api.expect(login_details)
    @api.marshal_with(login_details, envelope='the_data',mask='pass_word')
    def post(self):
        login_details = api.payload
        print(login_details) …

我正在尝试运行一个查询，根据字典列表过滤结果。

让我们更清楚...

我有一个表标题，每个标题都有一个版本字段

class Heading(Base):
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    version = Column(Integer)

另一方面，我有一个看起来像这样的字典列表（Json 格式）：

"headings": [
    {
        "id": 1,
        "version": 1
    },
    {
        "id": 2,
        "version": 1
    }
]

我想在 Heading 表上执行一个查询，排除数据库中的版本等于 (json) 列表中的版本的行。

我尝试执行这样的事情：

res = Heading.query.filter(~Heading.id.in_([d['id'] for d in headings if d['version'] == Heading.version])).all()

数据库看起来像这样（再次格式化 Json，抱歉）：

[
{
    "id": 1,
    "name": "Heading1",
    "version": 2,
},
{
    "id": 2,
    "name": "Heading2",
    "version": 1,
}
]

但无论如何它都会返回 Heading2 行。基本上，我希望查询返回 Json 中未定义的所有行以及数据库中具有较新版本的行。

我不知道我的问题是否足够清楚。也许这不是正确的方法，请告诉我

预先感谢您的任何帮助

编辑： 生成的原始 SQL …