- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
好的:我正在使用 Flask-SQLAlchemy 和 PostgreSQL 构建一个 Flask 应用程序,我想在我的数据库中创建一个可变的 JSON 类型列。 SQLAlchemy 中有很多在线示例(以及这里有很多关于自定义可变对象类型的问题)。我找到了这个:sqlalchemy-json (完整的文章 by the author here )处理可变的 JSON 对象类型。从理论上讲,它提供了 JsonObject,一种 JSON 对象类型,在基本级别对 dict 和列表进行更改跟踪,以及 NestedJsonObject,一种 JSON 对象类型,对 dict 和列表进行嵌套更改跟踪。它看起来很甜蜜。
我一生都无法让它发挥作用,尽管我完全有信心。请注意:上述作者的文章没有显示在列中实现对象类型的示例,因此由于我是新手,我很可能将下一部分弄错了。不过我咨询了sqlalchemy.ext.mutable reference这似乎是对的。
这是我的models.py:
from application import app
from flask_sqlalchemy import SQLAlchemy
from flask_login import UserMixin
import sqlalchemy_json # I'm using Alembic for migrations and make this import in my script.py.mako, too, in case that matters
from sqlalchemy_json import NestedJsonObject
db = SQLAlchemy(app)
class User(UserMixin, db.Model): #flask-sqlalchemy provides a Base declaration with db.Model
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(80), unique=True)
data = db.Column(NestedJsonObject)
def __init__(self, name, data):
self.name = name
self.data = data
def __repr__(self):
return "<User(id='%d', name='%s', connections='%r')>" % (self.id, self.name, self.data)
...
data = {}
data['a']= {'b': 'c', 'd': 'e'}
user = User( ... , data)
db.session.add(user)
db.session.commit()
#<User( ... data='{u'a': {u'b': u'c', u'd':u'e'}}')>
user = db.session.query(User).filter( ... ).first()
user.data['foo']={}
db.session.commit()
#<User( ... data='{u'a': {u'b': u'c', u'd':u'e'}}')>
user.data['foo']['bar'] = {'x': 'x', 'z': 'z'}
db.session.commit()
#<User( ... data='{u'a': {u'b': u'c', u'd':u'e'}}')>
.associate_with
位于文件底部:
# Third-party modules
try:
import simplejson as json
except ImportError:
import json
import sqlalchemy
from sqlalchemy.ext import mutable
# Custom modules
from . import track
class NestedMutable(mutable.MutableDict, track.TrackedDict):
"""SQLAlchemy `mutable` extension dictionary with nested change tracking."""
def __setitem__(self, key, value):
"""Ensure that items set are converted to change-tracking types."""
super(NestedMutable, self).__setitem__(key, self.convert(value, self))
@classmethod
def coerce(cls, key, value):
"""Convert plain dictionary to NestedMutable."""
if isinstance(value, cls):
return value
if isinstance(value, dict):
return cls(value)
return super(cls).coerce(key, value)
class _JsonTypeDecorator(sqlalchemy.TypeDecorator):
"""Enables JSON storage by encoding and decoding on the fly."""
impl = sqlalchemy.String
def process_bind_param(self, value, dialect):
return json.dumps(value)
def process_result_value(self, value, dialect):
return json.loads(value)
class JsonObject(_JsonTypeDecorator):
"""JSON object type for SQLAlchemy with change tracking as base level."""
class NestedJsonObject(_JsonTypeDecorator):
"""JSON object type for SQLAlchemy with nested change tracking."""
mutable.MutableDict.associate_with(JsonObject)
NestedMutable.associate_with(NestedJsonObject)
#!/usr/bin/python
"""This module contains the tracked object classes.
TrackedObject forms the basis for both the TrackedDict and the TrackedList.
A function for automatic conversion of dicts and lists to their tracked
counterparts is also included.
"""
# Standard modules
import itertools
import logging
class TrackedObject(object):
"""A base class for delegated change-tracking."""
_type_mapping = {}
def __init__(self, *args, **kwds):
self.logger = logging.getLogger(type(self).__name__)
self.logger.debug('%s: __init__', self._repr())
self.parent = None
super(TrackedObject, self).__init__(*args, **kwds)
def changed(self, message=None, *args):
"""Marks the object as changed.
If a `parent` attribute is set, the `changed()` method on the parent will
be called, propagating the change notification up the chain.
The message (if provided) will be debug logged.
"""
if message is not None:
self.logger.debug('%s: %s', self._repr(), message % args)
self.logger.debug('%s: changed', self._repr())
if self.parent is not None:
self.parent.changed()
@classmethod
def register(cls, origin_type):
"""Registers the class decorated with this method as a mutation tracker.
The provided `origin_type` is mapped to the decorated class such that
future calls to `convert()` will convert the object of `origin_type` to an
instance of the decorated class.
"""
def decorator(tracked_type):
"""Adds the decorated class to the `_type_mapping` dictionary."""
cls._type_mapping[origin_type] = tracked_type
return tracked_type
return decorator
@classmethod
def convert(cls, obj, parent):
"""Converts objects to registered tracked types
This checks the type of the given object against the registered tracked
types. When a match is found, the given object will be converted to the
tracked type, its parent set to the provided parent, and returned.
If its type does not occur in the registered types mapping, the object
is returned unchanged.
"""
obj_type = type(obj)
for origin_type, replacement in cls._type_mapping.iteritems():
if obj_type is origin_type:
new = replacement(obj)
new.parent = parent
return new
return obj
@classmethod
def convert_iterable(cls, iterable, parent):
"""Returns a generator that performs `convert` on every of its members."""
return (cls.convert(item, parent) for item in iterable)
@classmethod
def convert_iteritems(cls, iteritems, parent):
"""Returns a generator like `convert_iterable` for 2-tuple iterators."""
return ((key, cls.convert(value, parent)) for key, value in iteritems)
@classmethod
def convert_mapping(cls, mapping, parent):
"""Convenience method to track either a dict or a 2-tuple iterator."""
if isinstance(mapping, dict):
return cls.convert_iteritems(mapping.iteritems(), parent)
return cls.convert_iteritems(mapping, parent)
def _repr(self):
"""Simple object representation."""
return '<%(namespace)s.%(type)s object at 0x%(address)0xd>' % {
'namespace': __name__,
'type': type(self).__name__,
'address': id(self)}
@TrackedObject.register(dict)
class TrackedDict(TrackedObject, dict):
"""A TrackedObject implementation of the basic dictionary."""
def __init__(self, source=(), **kwds):
super(TrackedDict, self).__init__(itertools.chain(
self.convert_mapping(source, self),
self.convert_mapping(kwds, self)))
def __setitem__(self, key, value):
self.changed('__setitem__: %r=%r', key, value)
super(TrackedDict, self).__setitem__(key, self.convert(value, self))
def __delitem__(self, key):
self.changed('__delitem__: %r', key)
super(TrackedDict, self).__delitem__(key)
def clear(self):
self.changed('clear')
super(TrackedDict, self).clear()
def pop(self, *key_and_default):
self.changed('pop: %r', key_and_default)
return super(TrackedDict, self).pop(*key_and_default)
def popitem(self):
self.changed('popitem')
return super(TrackedDict, self).popitem()
def update(self, source=(), **kwds):
self.changed('update(%r, %r)', source, kwds)
super(TrackedDict, self).update(itertools.chain(
self.convert_mapping(source, self),
self.convert_mapping(kwds, self)))
@TrackedObject.register(list)
class TrackedList(TrackedObject, list):
"""A TrackedObject implementation of the basic list."""
def __init__(self, iterable=()):
super(TrackedList, self).__init__(self.convert_iterable(iterable, self))
def __setitem__(self, key, value):
self.changed('__setitem__: %r=%r', key, value)
super(TrackedList, self).__setitem__(key, self.convert(value, self))
def __delitem__(self, key):
self.changed('__delitem__: %r', key)
super(TrackedList, self).__delitem__(key)
def append(self, item):
self.changed('append: %r', item)
super(TrackedList, self).append(self.convert(item, self))
def extend(self, iterable):
self.changed('extend: %r', iterable)
super(TrackedList, self).extend(self.convert_iterable(iterable, self))
def remove(self, value):
self.changed('remove: %r', value)
return super(TrackedList, self).remove(value)
def pop(self, index):
self.changed('pop: %d', index)
return super(TrackedList, self).pop(index)
def sort(self, cmp=None, key=None, reverse=False):
self.changed('sort')
super(TrackedList, self).sort(cmp=cmp, key=key, reverse=reverse)
from .alchemy import NestedJsonObject, NestedMutable, JsonObject
__all__ = (
'NestedJsonObject',
'NestedMutable',
'JsonObject'
)
Copyright (c) 2014, Elmer de Looff <elmer.delooff@gmail.com>
All rights reserved.
Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:
1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer.
2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
最佳答案
上网查了一下,sqlalchemy-json(感谢edelooff)是最好的选择,但是origin one只支持mutable dict,this fork由 torotil 更新解决了问题,感谢 torotil!
并将记录器放在 TrackedObject 中会导致深度复制问题,最好移到模块中。我 fork 了一个新的存储库来解决这个问题:Cysnake4713
另外,我使用 origin sqlalchemy.sql.sqltypes.JSON 而不是自定义的 JsonObject,看起来效果很好。
只需将 associate_with 部分替换为:
NestedMutable.associate_with(db.JSON)
关于json - 不跟踪更改的可变 JSON 对象类型 SQLAlchemy (sqlalchemy-json),我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/39782773/
我正在尝试创建一个使用 UUID 作为主键的用户模型: from src.db import db # SQLAlchemy instance import sqlalchemy_utils impo
在 sqlalchemy 中,我试图合并表,然后使用 WHERE 和 ORDER_BY 创建别名 有点像 SELECT * FROM ( SELECT [TABLE_ONE].[SOME_ID]
我正在使用 SQL Alchemy(通过 Flask_sqlalchemy)将 Python 字典列表插入到 Postgres 数据库中。 其中一个表是所有唯一项目的列表(表 1),而第二个是与某个项
This source详细说明如何使用关联代理创建具有 ORM 对象值的 View 和对象。 但是,当我附加一个与数据库中现有对象匹配的值(并且所述值是唯一的或主键)时,它会创建一个冲突的对象,因此我
SQLAlchemy Core和SQLAlchemy ORM的目的有什么区别? 最佳答案 顾名思义,ORM是一个对象关系映射器:其目的是将数据库关系表示为Python对象。 核心是查询构建器。其目的是
带有ForeignKey的Column是否自动创建索引? 还是我需要手动添加index=True? some_field = Column(Integer, ForeignKey(SomeModel.
我有一个主数据库,每个客户自己的数据库连接存储在其中。 因此,每个客户端都使用2个db:main和它自己的db,必须确定其连接 对于每个http调用。我如何使用flask-sqlalchemy扩展名执
当我仅对类进行继承时,它才起作用 class User(Base): __tablename__ = ’users’ id = Column(Integer, primary_key=
从用户的角度来看,SQLAlchemy 的查询日志似乎有点过于冗长,有时甚至有点神秘: 2015-10-02 13:51:39,500 INFO sqlalchemy.engine.base.Engi
我正在尝试使用 wtforms.ext.sqlalchemy QuerySelectMultipleField 显示复选框列表,但我无法在 GET 的表单上显示模型数据。 这是我的models.py
我想为查询返回一个中继连接。使用标准的 graphene-sqlalchemy 你可以这样做: class Query(graphene.ObjectType): node = relay.N
我在 centos 7.5 虚拟机上部署了最新的 Airflow ,并将 sql_alchemy_conn 和 result_backend 更新到 postgresql 实例上的 postgres
我想将多个项目插入到一个表中,并在发生冲突时更新该表。这是我想出的以下内容 from sqlalchemy.dialects.postgresql import insert meta = MetaD
我有以下模型: class Item(Base): a = relationship() b = relationship() c = relationship() d
我有 presto 和 superset 设置。 presto 运行良好,可以通过命令访问: ./app/hadoop/setjdk8.sh;bin/presto-cli --server http:
我一直在寻找一种在 sqlalchemy 中使用 tsvector 的方法(就像 INTEGER 等其他方法一样),但到目前为止我还不清楚如何做到这一点。我读过可以使用 UserDefinedType
我正在使用 sqlalchemy(现在使用 sqlite,但稍后可能会改变)来构建一个数据库,其中插入的顺序和 rowids 很重要。我基本上有以下几点: class Message(Base):
给定一个对象,我想知道如何知道它是否是 sqlalchemy 映射模型的实例。 通常,我会使用 isinstance(obj, DeclarativeBase)。但是,在这种情况下,我没有可用的 De
我已经通读了查询文档,如果有办法从查询中获取表名,就看不到任何地方 - 例如如果我有 q = query(Users) ,我可以得到Users从 q 退出? 最佳答案 请注意,像您这样的事件简单查询可
我不确定如何定义create schema foo迁移?我的模型如下所示(我正在使用Flask-Migrate): class MyTable(db.Model): __tablename__
我是一名优秀的程序员,十分优秀!