##// END OF EJS Templates
api: use consistent way to extract users, repos, repo groups and user groups by id or name....
api: use consistent way to extract users, repos, repo groups and user groups by id or name. - makes usage of Number vs String to differenciate if we pick objec ID or it's name this will allow easy fetching of objects by either id or it's name, including numeric string name - fixes #5230

File last commit:

r1271:47a44c03 default
r1530:1efcb4ee default
Show More
jsonalchemy.py
265 lines | 8.4 KiB | text/x-python | PythonLexer
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2017 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import collections
import sqlalchemy
from sqlalchemy import UnicodeText
from sqlalchemy.ext.mutable import Mutable
from rhodecode.lib.ext_json import json
class JsonRaw(unicode):
"""
Allows interacting with a JSON types field using a raw string.
For example::
db_instance = JsonTable()
db_instance.enabled = True
db_instance.json_data = JsonRaw('{"a": 4}')
This will bypass serialization/checks, and allow storing
raw values
"""
pass
# Set this to the standard dict if Order is not required
DictClass = collections.OrderedDict
class JSONEncodedObj(sqlalchemy.types.TypeDecorator):
"""
Represents an immutable structure as a json-encoded string.
If default is, for example, a dict, then a NULL value in the
database will be exposed as an empty dict.
"""
impl = UnicodeText
safe = True
def __init__(self, *args, **kwargs):
self.default = kwargs.pop('default', None)
self.safe = kwargs.pop('safe_json', self.safe)
self.dialect_map = kwargs.pop('dialect_map', {})
super(JSONEncodedObj, self).__init__(*args, **kwargs)
def load_dialect_impl(self, dialect):
if dialect.name in self.dialect_map:
return dialect.type_descriptor(self.dialect_map[dialect.name])
return dialect.type_descriptor(self.impl)
def process_bind_param(self, value, dialect):
if isinstance(value, JsonRaw):
value = value
elif value is not None:
value = json.dumps(value)
return value
def process_result_value(self, value, dialect):
if self.default is not None and (not value or value == '""'):
return self.default()
if value is not None:
try:
value = json.loads(value, object_pairs_hook=DictClass)
except Exception as e:
if self.safe:
return self.default()
else:
raise
return value
class MutationObj(Mutable):
@classmethod
def coerce(cls, key, value):
if isinstance(value, dict) and not isinstance(value, MutationDict):
return MutationDict.coerce(key, value)
if isinstance(value, list) and not isinstance(value, MutationList):
return MutationList.coerce(key, value)
return value
@classmethod
def _listen_on_attribute(cls, attribute, coerce, parent_cls):
key = attribute.key
if parent_cls is not attribute.class_:
return
# rely on "propagate" here
parent_cls = attribute.class_
def load(state, *args):
val = state.dict.get(key, None)
if coerce:
val = cls.coerce(key, val)
state.dict[key] = val
if isinstance(val, cls):
val._parents[state.obj()] = key
def set(target, value, oldvalue, initiator):
if not isinstance(value, cls):
value = cls.coerce(key, value)
if isinstance(value, cls):
value._parents[target.obj()] = key
if isinstance(oldvalue, cls):
oldvalue._parents.pop(target.obj(), None)
return value
def pickle(state, state_dict):
val = state.dict.get(key, None)
if isinstance(val, cls):
if 'ext.mutable.values' not in state_dict:
state_dict['ext.mutable.values'] = []
state_dict['ext.mutable.values'].append(val)
def unpickle(state, state_dict):
if 'ext.mutable.values' in state_dict:
for val in state_dict['ext.mutable.values']:
val._parents[state.obj()] = key
sqlalchemy.event.listen(parent_cls, 'load', load, raw=True,
propagate=True)
sqlalchemy.event.listen(parent_cls, 'refresh', load, raw=True,
propagate=True)
sqlalchemy.event.listen(parent_cls, 'pickle', pickle, raw=True,
propagate=True)
sqlalchemy.event.listen(attribute, 'set', set, raw=True, retval=True,
propagate=True)
sqlalchemy.event.listen(parent_cls, 'unpickle', unpickle, raw=True,
propagate=True)
class MutationDict(MutationObj, DictClass):
@classmethod
def coerce(cls, key, value):
"""Convert plain dictionary to MutationDict"""
self = MutationDict(
(k, MutationObj.coerce(key, v)) for (k, v) in value.items())
self._key = key
return self
def __setitem__(self, key, value):
# Due to the way OrderedDict works, this is called during __init__.
# At this time we don't have a key set, but what is more, the value
# being set has already been coerced. So special case this and skip.
if hasattr(self, '_key'):
value = MutationObj.coerce(self._key, value)
DictClass.__setitem__(self, key, value)
self.changed()
def __delitem__(self, key):
DictClass.__delitem__(self, key)
self.changed()
def __setstate__(self, state):
self.__dict__ = state
def __reduce_ex__(self, proto):
# support pickling of MutationDicts
d = dict(self)
return (self.__class__, (d, ))
class MutationList(MutationObj, list):
@classmethod
def coerce(cls, key, value):
"""Convert plain list to MutationList"""
self = MutationList((MutationObj.coerce(key, v) for v in value))
self._key = key
return self
def __setitem__(self, idx, value):
list.__setitem__(self, idx, MutationObj.coerce(self._key, value))
self.changed()
def __setslice__(self, start, stop, values):
list.__setslice__(self, start, stop,
(MutationObj.coerce(self._key, v) for v in values))
self.changed()
def __delitem__(self, idx):
list.__delitem__(self, idx)
self.changed()
def __delslice__(self, start, stop):
list.__delslice__(self, start, stop)
self.changed()
def append(self, value):
list.append(self, MutationObj.coerce(self._key, value))
self.changed()
def insert(self, idx, value):
list.insert(self, idx, MutationObj.coerce(self._key, value))
self.changed()
def extend(self, values):
list.extend(self, (MutationObj.coerce(self._key, v) for v in values))
self.changed()
def pop(self, *args, **kw):
value = list.pop(self, *args, **kw)
self.changed()
return value
def remove(self, value):
list.remove(self, value)
self.changed()
def JsonType(impl=None, **kwargs):
"""
Helper for using a mutation obj, it allows to use .with_variant easily.
example::
settings = Column('settings_json',
MutationObj.as_mutable(
JsonType(dialect_map=dict(mysql=UnicodeText(16384))))
"""
if impl == 'list':
return JSONEncodedObj(default=list, **kwargs)
elif impl == 'dict':
return JSONEncodedObj(default=DictClass, **kwargs)
else:
return JSONEncodedObj(**kwargs)
JSON = MutationObj.as_mutable(JsonType())
"""
A type to encode/decode JSON on the fly
sqltype is the string type for the underlying DB column::
Column(JSON) (defaults to UnicodeText)
"""
JSONDict = MutationObj.as_mutable(JsonType('dict'))
"""
A type to encode/decode JSON dictionaries on the fly
"""
JSONList = MutationObj.as_mutable(JsonType('list'))
"""
A type to encode/decode JSON lists` on the fly
"""