本文整理汇总了Python中sqlalchemy.ext.mutable.MutableDict类的典型用法代码示例。如果您正苦于以下问题:Python MutableDict类的具体用法?Python MutableDict怎么用?Python MutableDict使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了MutableDict类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: setup_mappers
def setup_mappers(cls):
foo = cls.tables.foo
subfoo = cls.tables.subfoo
mapper(Foo, foo)
mapper(SubFoo, subfoo, inherits=Foo)
MutableDict.associate_with_attribute(Foo.data)
开发者ID:Daniel-B-Smith,项目名称:sqlalchemy,代码行数:7,代码来源:test_mutable.py
示例2: define_tables
def define_tables(cls, metadata):
import json
class JSONEncodedDict(TypeDecorator):
impl = VARCHAR(50)
def process_bind_param(self, value, dialect):
if value is not None:
value = json.dumps(value)
return value
def process_result_value(self, value, dialect):
if value is not None:
value = json.loads(value)
return value
MutableDict = cls._type_fixture()
MutableDict.associate_with(JSONEncodedDict)
Table('foo', metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('data', JSONEncodedDict),
Column('unrelated_data', String(50))
)
开发者ID:Daniel-B-Smith,项目名称:sqlalchemy,代码行数:26,代码来源:test_mutable.py
示例3: upgrade
def upgrade():
meta_catalogstar = sa.Column('meta', MutableDict.as_mutable(JSON),
default={})
op.add_column('catalog_star', meta_catalogstar)
meta_obs = sa.Column('meta', MutableDict.as_mutable(JSON),
default={})
op.add_column('observation', meta_obs)
开发者ID:jonathansick,项目名称:starplex,代码行数:8,代码来源:effa43e7773_add_hstore_to_catalogstar_observation.py
示例4: associate_with
def associate_with(sqltype):
# TODO(leizhang) When we removed sqlalchemy 0.7 dependence
# we can import MutableDict directly and remove ./mutable.py
try:
from sqlalchemy.ext.mutable import MutableDict as sa_MutableDict
sa_MutableDict.associate_with(Json)
except ImportError:
from heat.db.sqlalchemy.mutable import MutableDict
MutableDict.associate_with(Json)
开发者ID:andrew-plunk,项目名称:heat,代码行数:9,代码来源:types.py
示例5: document
def document(self):
d = MutableDict()
d.update(self._document)
d.update({
'kind': self.kind,
'id': self.id,
'source': self.source,
'timestamp': self.timestamp,
})
return d
开发者ID:dhruvbaldawa,项目名称:vella,代码行数:10,代码来源:postgresql.py
示例6: downgrade
def downgrade():
op.drop_column('catalog', 'metajson')
meta = sa.Column('meta', MutableDict.as_mutable(HSTORE),
nullable=False,
default={},
index=True)
op.add_column('catalog', meta)
开发者ID:jonathansick,项目名称:starplex,代码行数:7,代码来源:595ab4e89b02_add_catalog_metajson_column.py
示例7: define_tables
def define_tables(cls, metadata):
import json
class JSONEncodedDict(TypeDecorator):
impl = VARCHAR(50)
def process_bind_param(self, value, dialect):
if value is not None:
value = json.dumps(value)
return value
def process_result_value(self, value, dialect):
if value is not None:
value = json.loads(value)
return value
MutableDict = cls._type_fixture()
Table(
"foo",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("data", MutableDict.as_mutable(JSONEncodedDict)),
Column("non_mutable_data", JSONEncodedDict),
Column("unrelated_data", String(50)),
)
开发者ID:BY-jk,项目名称:sqlalchemy,代码行数:29,代码来源:test_mutable.py
示例8: define_tables
def define_tables(cls, metadata):
MutableDict = cls._type_fixture()
mutable_pickle = MutableDict.as_mutable(PickleType)
Table(
'foo', metadata,
Column(
'id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('data', mutable_pickle, default={}),
)
开发者ID:gencer,项目名称:sqlalchemy,代码行数:11,代码来源:test_mutable.py
示例9: __init__
def __init__(self, name=u"", type=u"node", id=None, schema=None, attrs=None, system_attrs=None, orderpos=None):
self.name = name
if not isinstance(type, unicode):
warn("type arg of Node should be unicode (hint: don't create nodes with Node(type='{}')!)".format(type), DeprecationWarning)
if "/" in type:
warn("use separate type and schema parameters instead of 'type/schema'", DeprecationWarning)
type, schema = type.split("/")
self.type = type
self.attrs = MutableDict()
self.system_attrs = MutableDict()
if id:
self.id = id
if schema:
self.schema = schema
if attrs:
self.attrs.update(attrs)
if system_attrs:
self.system_attrs.update(system_attrs)
if orderpos:
self.orderpos = orderpos
开发者ID:mediatum,项目名称:mediatum,代码行数:22,代码来源:node.py
示例10: cls
return cls(value)
return super(cls).coerce(key, value)
class NestedMutable(Mutable):
"""SQLAlchemy `mutable` extension with nested change tracking."""
@classmethod
def coerce(cls, key, value):
"""Convert plain dictionary to NestedMutable."""
if value is None:
return value
if isinstance(value, cls):
return value
if isinstance(value, dict):
return NestedMutableDict.coerce(key, value)
if isinstance(value, list):
return NestedMutableList.coerce(key, value)
return super(cls).coerce(key, value)
class MutableJson(JSONType):
"""JSON type for SQLAlchemy with change tracking at top level."""
class NestedMutableJson(JSONType):
"""JSON type for SQLAlchemy with nested change tracking."""
MutableDict.associate_with(MutableJson)
NestedMutable.associate_with(NestedMutableJson)
开发者ID:edelooff,项目名称:sqlalchemy-json,代码行数:30,代码来源:__init__.py
示例11: Column
@author: peterb
'''
from blueshed.model_helpers.sqla_views import view
from blueshed.model_helpers.sql_extensions import JSONEncodedDict
from sqlalchemy.ext.mutable import MutableDict
from sqlalchemy.sql.expression import select, join
from sqlalchemy.sql.functions import func
from examples.simple.model import *
from blueshed.model_helpers.access_model import Person, Permission,\
person_permissions_permission
Person._token = Column(String(80))
Person._preferences = Column(MutableDict.as_mutable(JSONEncodedDict(255)))
Person.firstname = Column(String(80))
Person.lastname = Column(String(80))
Person.photo = Column(String(128))
'''
An example View
'''
q = select([Person.id.label('id'),
Person.email.label('email'),
func.count(Permission.id).label('permission_count')]).\
select_from(join(Person,
person_permissions_permission,
Person.id==person_permissions_permission.c.permissions_id).\
join(Permission,
Permission.id==person_permissions_permission.c.permission_id)).\
开发者ID:blueshed,项目名称:blueshed-py,代码行数:30,代码来源:model_ext.py
示例12: Node
class Node(DeclarativeBase, NodeMixin):
"""Base class for Nodes which holds all SQLAlchemy fields definitions
"""
__metaclass__ = BaseNodeMeta
__tablename__ = "node"
__versioned__ = {
"base_classes": (NodeVersionMixin, MtVersionBase, DeclarativeBase),
"exclude": ["subnode", "system_attrs"]
}
id = C(Integer, node_id_seq, server_default=node_id_seq.next_value(), primary_key=True)
type = C(Unicode, index=True)
schema = C(Unicode, index=True)
name = C(Unicode, index=True)
orderpos = C(Integer, default=1, index=True)
fulltext = deferred(C(Unicode))
# indicate that this node is a subnode of a content type node
# subnode exists just for performance reasons and is updated by the database
# unversioned
subnode = C(Boolean, server_default="false")
attrs = deferred(C(MutableDict.as_mutable(JSONB)))
# Migration from old mediatum: all attributes starting with "system." go here.
# We should get rid of most (all?) such attributes in the future.
# unversioned
system_attrs = deferred(C(MutableDict.as_mutable(JSONB)))
@hybrid_property
def a_expr(self):
""" see: Attributes"""
raise Exception("node.a_expr")
if "_attributes_accessor" not in self.__dict__:
setattr(self, "_attributes_accessor", Attributes(self, "attrs"))
return self._attributes_accessor
@a_expr.expression
def a(self):
""" see: AttributesExpression"""
if "_attributes_accessor" not in self.__dict__:
setattr(self, "_attributes_accessor", AttributesExpressionAdapter(self, "attrs"))
return self._attributes_accessor
@a.setter
def a_set(self, value):
raise NotImplementedError("immutable!")
@hybrid_property
def sys(self):
""" see: Attributes"""
if "_system_attributes_accessor" not in self.__dict__:
setattr(self, "_system_attributes_accessor", Attributes(self, "system_attrs"))
return self._system_attributes_accessor
@sys.expression
def sys_expr(self):
""" see: AttributesExpression"""
if "_system_attributes_accessor" not in self.__dict__:
setattr(self, "_system_attributes_accessor", AttributesExpressionAdapter(self, "system_attrs"))
return self._system_attributes_accessor
@a.setter
def sys_set(self, value):
raise NotImplementedError("immutable!")
def __init__(self, name=u"", type=u"node", id=None, schema=None, attrs=None, system_attrs=None, orderpos=None):
self.name = name
if not isinstance(type, unicode):
warn("type arg of Node should be unicode (hint: don't create nodes with Node(type='{}')!)".format(type), DeprecationWarning)
if "/" in type:
warn("use separate type and schema parameters instead of 'type/schema'", DeprecationWarning)
type, schema = type.split("/")
self.type = type
self.attrs = MutableDict()
self.system_attrs = MutableDict()
if id:
self.id = id
if schema:
self.schema = schema
if attrs:
self.attrs.update(attrs)
if system_attrs:
self.system_attrs.update(system_attrs)
if orderpos:
self.orderpos = orderpos
@property
def slow_content_children_for_all_subcontainers(self):
"""
!!! very slow, use content_children_for_all_subcontainers instead!!!
Collects all Content nodes in all subcontainers of this node.
This excludes content nodes that are children of other content nodes.
"""
warn("very slow, use content_children_for_all_subcontainers instead", DeprecationWarning)
from contenttypes.data import Content
from core import db
#.........这里部分代码省略.........
开发者ID:mediatum,项目名称:mediatum,代码行数:101,代码来源:node.py
示例13: AstonFrameBinary
return json.loads(value)
class AstonFrameBinary(TypeDecorator):
impl = LargeBinary
def process_bind_param(self, value, dialect):
if value is not None:
return value.compress()
def process_result_value(self, value, dialect):
if value is not None:
return decompress(value)
MutableDict.associate_with(JSONDict)
Base = declarative_base()
def initialize_sql(engine):
DBSession = scoped_session(sessionmaker(expire_on_commit=False))
DBSession.configure(bind=engine)
Base.metadata.bind = engine
Base.metadata.create_all(engine)
return DBSession
def quick_sqlite(filename):
from sqlalchemy import create_engine
开发者ID:bovee,项目名称:AstonQt,代码行数:30,代码来源:__init__.py
示例14: process_bind_param
impl = HSTORE
def process_bind_param(self, value, dialect):
if not isinstance(value, dict):
return value
ret = {k: json.dumps(v) for k, v in value.items()}
return ret
def process_result_value(self, value, dialect):
if not value:
return MutableDict()
ret = MutableDict({k: json.loads(v) for k, v in value.items()})
return ret
MutableDict.associate_with(JSONValuesColumn)
class IntegerEnumColumn(types.TypeDecorator):
impl = types.INTEGER
def __init__(self, enum_values):
super(IntegerEnumColumn, self).__init__()
self.enum_values = enum_values
self.reverse_enum_values = reverse_dict(enum_values)
def process_bind_param(self, value, dialect):
return self.reverse_enum_values.get(value, value)
def process_result_value(self, value, dialect):
return self.enum_values.get(value, value)
开发者ID:dzamie,项目名称:weasyl,代码行数:31,代码来源:helpers.py
示例15: array_base
#XXX NOTICE XXX DO NOT NAME THINGS types.py it breaks EVERYTHING
from sqlalchemy.types import PickleType
from sqlalchemy.dialects import postgres,postgresql
from sqlalchemy.ext.mutable import MutableDict
def array_base(column_type):
array = PickleType()
array.with_variant(postgres.ARRAY(column_type), 'postgres')
array.with_variant(postgres.ARRAY(column_type), 'postgresql')
return array
Array=array_base
_DictType = MutableDict.as_mutable(PickleType)
_DictType.with_variant(MutableDict.as_mutable(postgres.HSTORE), 'postgres')
#_DictType.with_variant(MutableDict.as_mutable(postgresql.HSTORE), 'postgresql')
#_DictType.with_variant(MutableDict.as_mutable(postgresql.HSTORE), 'psycopg2')
#_DictType.with_variant(MutableDict.as_mutable(postgresql.HSTORE), 'postgresql+psycopg2')
DictType=_DictType #FIXME not working as hstore :/
__all__=[
'Array',
'DictType',
]
#ArrayFloat = PickleType()
#ArrayFloat.with_variant(postgresql.ARRAY(Float), 'postgresql')
#ArrayString = PickleType()
#ArrayString.with_variant(postgresql.ARRAY(String), 'postgresql')
开发者ID:tgbugs,项目名称:mlab,代码行数:31,代码来源:types_.py
示例16: process_bind_param
impl = VARCHAR
def process_bind_param(self, value, dialect):
if value is not None:
value = str(value)
return value
def process_result_value(self, value, dialect):
if value is not None:
value = ast.literal_eval(value)
return value
ARRAY_TYPE = Array()
JSON_TYPE = MutableDict.as_mutable(JSONEncodedDict)
BASE = declarative_base()
class Product(BASE):
"""docstring for Product."""
__tablename__ = "products"
id = Column('id', Integer, primary_key=True, autoincrement=False)
style_no = Column('style_no', String)
title = Column('title', String)
keywords = Column('keywords', ARRAY_TYPE)
owner = Column('owner', String)
modify_time = Column('modify_time', Date)
update = Column('update', Date, default=date.today)
开发者ID:everyx,项目名称:AliTools,代码行数:30,代码来源:models.py
示例17: process_bind_param
_null = None
_type = object
def process_bind_param(self, value, dialect):
return json.dumps(value)
def process_literal_param(self, value, dialect):
return value
def process_result_value(self, value, dialect):
try:
value = json.loads(value)
except (ValueError, TypeError):
value = self._null
return value
class List(Json):
_null = []
_type = list
class Dict(Json):
_null = {}
_type = dict
MutableDict.associate_with(Dict)
Base = declarative_base()
开发者ID:meizhuhanxiang,项目名称:preseller,代码行数:30,代码来源:base.py
示例18: get_data_element_postgres_extensions
def get_data_element_postgres_extensions(task, indexes):
if indexes:
q = task.__class__.data[indexes]
else:
q = task.__class__.data
s = object_session(task)
tup = s.query(q).filter_by(id=task.id).one()
return tup[0]
class json_array_length(GenericFunction):
type = Integer
def get_data_size_postgres_extensions(task, indexes):
if indexes:
q = task.__class__.data[indexes]
else:
q = task.__class__.data
s = object_session(task)
tup = s.query(json_array_length(q)).filter_by(id=task.id).one()
return tup[0]
MutableJSONDict = MutableDict.as_mutable(psqlJSON)
JSON = psqlJSON
get_data_element = get_data_element_postgres_extensions
get_data_size = get_data_size_postgres_extensions
开发者ID:davidlmorton,项目名称:ptero-workflow,代码行数:29,代码来源:json_type.py
示例19: hstore_table_for
def hstore_table_for(self, name):
return sa.Table(name, self.Base.metadata,
sa.Column('id', sa.Integer, primary_key=True),
sa.Column('data', MutableDict.as_mutable(pg.HSTORE)))
开发者ID:kvikshaug,项目名称:grouphugs-py,代码行数:4,代码来源:database.py
示例20: Json
class Json(types.TypeDecorator):
impl = types.Text
def process_bind_param(self, value, dialect):
return dumps(value)
def process_result_value(self, value, dialect):
return loads(value)
# TODO(leizhang) When we removed sqlalchemy 0.7 dependence
# we can import MutableDict directly and remove ./mutable.py
try:
from sqlalchemy.ext.mutable import MutableDict as sa_MutableDict
sa_MutableDict.associate_with(Json)
except ImportError:
from heat.db.sqlalchemy.mutable import MutableDict
MutableDict.associate_with(Json)
class HeatBase(object):
"""Base class for Heat Models."""
__table_args__ = {'mysql_engine': 'InnoDB'}
__table_initialized__ = False
created_at = sqlalchemy.Column(sqlalchemy.DateTime,
default=timeutils.utcnow)
updated_at = sqlalchemy.Column(sqlalchemy.DateTime,
onupdate=timeutils.utcnow)
def save(self, session=None):
开发者ID:jake-liu,项目名称:heat,代码行数:29,代码来源:models.py
注:本文中的sqlalchemy.ext.mutable.MutableDict类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论