Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _readonly method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

json_wrappers.py

Source:json_wrappers.py

1# @license2# Copyright 2017 Google Inc.3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7#      http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14"""Facilities for converting JSON <-> Python objects"""15from __future__ import absolute_import16import collections17import copy18import inspect19import numbers20import threading21import numpy as np22import six23from six.moves import range24from .json_utils import encode_json_for_repr25def to_json(value):26    if isinstance(value, np.ndarray):27        return value.tolist()28    try:29        method = value.to_json30    except AttributeError:31        return value32    return method()33class JsonObjectWrapper(object):34    supports_readonly = True35    __slots__ = ('_json_data', '_cached_wrappers', '_lock', '_readonly')36    def __init__(self, json_data=None, _readonly=False, **kwargs):37        if json_data is None:38            json_data = collections.OrderedDict()39        elif isinstance(json_data, type(self)):40            json_data = json_data.to_json()41        elif not isinstance(json_data, dict):42            raise TypeError43        object.__setattr__(self, '_json_data', json_data)44        object.__setattr__(self, '_cached_wrappers', dict())45        object.__setattr__(self, '_lock', threading.RLock())46        object.__setattr__(self, '_readonly', 1 if _readonly else False)47        for k in kwargs:48            setattr(self, k, kwargs[k])49        object.__setattr__(self, '_readonly', _readonly)50    def to_json(self):51        if self._readonly:52            return self._json_data53        with self._lock:54            r = self._json_data.copy()55            for k, (wrapper, _) in six.iteritems(self._cached_wrappers):56                if wrapper is not None:57                    r[k] = to_json(wrapper)58                else:59                    r.pop(k, None)60            return r61    def __deepcopy__(self, memo):62        return type(self)(copy.deepcopy(self.to_json(), memo))63    def __eq__(self, other):64        return type(self) == type(other) and self.to_json() == other.to_json()65    def __repr__(self):66        return u'%s(%s)' % (type(self).__name__, encode_json_for_repr(self.to_json()))67    def _get_wrapped(self, key, wrapped_type):68        with self._lock:69            json_value = self._json_data.get(key)70            cached_value = self._cached_wrappers.get(key)71            if cached_value is not None and cached_value[1] is json_value:72                return cached_value[0]73            kwargs = dict()74            if self._readonly and hasattr(wrapped_type, 'supports_readonly'):75                kwargs['_readonly'] = True76            wrapper = wrapped_type(json_value, **kwargs)77            self._cached_wrappers[key] = wrapper, json_value78            return wrapper79    def _set_wrapped(self, key, value, validator):80        if self._readonly is True:81            raise AttributeError82        value = validator(value)83        with self._lock:84            self._cached_wrappers[key] = (value, self._json_data.get(key))85_types_supporting_validation = frozenset([np.uint64, float, int])86def _normalize_validator(wrapped_type, validator):87    if validator is None:88        supports_validation = getattr(wrapped_type, 'supports_validation', None)89        if (inspect.isroutine(wrapped_type) or supports_validation is not None90                or wrapped_type in _types_supporting_validation):91            if inspect.isroutine(supports_validation):92                validator = supports_validation93            else:94                validator = wrapped_type95        else:96            def validator_func(x):97                if not isinstance(x, wrapped_type):98                    raise TypeError(wrapped_type, x)99                return x100            validator = validator_func101    return validator102def wrapped_property(json_name, wrapped_type, validator=None, doc=None):103    validator = _normalize_validator(wrapped_type, validator)104    return property(fget=lambda self: self._get_wrapped(json_name, wrapped_type),105                    fset=lambda self, value: self._set_wrapped(json_name, value, validator),106                    doc=doc)107def array_wrapper(dtype, shape=None):108    if shape is not None:109        if isinstance(shape, numbers.Number):110            shape = (shape, )111        else:112            shape = tuple(shape)113    def wrapper(value, _readonly=False):114        value = np.array(value, dtype=dtype)115        if _readonly:116            value.setflags(write=False)117        if shape is not None:118            if len(shape) != len(value.shape) or any(119                    expected_size is not None and expected_size != actual_size120                    for expected_size, actual_size in zip(shape, value.shape)):121                raise ValueError('expected shape', shape)122        return value123    wrapper.supports_readonly = True124    return wrapper125def text_type(value):126    return six.text_type(value)127def optional(wrapper, default_value=None, validator=None):128    def modified_wrapper(value, **kwargs):129        if value is None:130            return default_value131        return wrapper(value, **kwargs)132    if hasattr(wrapper, 'supports_readonly'):133        modified_wrapper.supports_readonly = True134    validator = _normalize_validator(wrapper, validator)135    def modified_validator(value, **kwargs):136        if value is None:137            return default_value138        return validator(value, **kwargs)139    modified_wrapper.supports_validation = modified_validator140    return modified_wrapper141class MapBase(object):142    __slots__ = ()143    pass144def typed_string_map(wrapped_type, validator=None):145    validator = _normalize_validator(wrapped_type, validator)146    class Map(JsonObjectWrapper, MapBase):147        supports_validation = True148        __slots__ = ()149        def __init__(self, json_data=None, _readonly=False):150            if isinstance(json_data, MapBase):151                json_data = json_data.to_json()152            elif json_data is not None:153                new_map = collections.OrderedDict()154                for k, v in six.viewitems(json_data):155                    validator(v)156                    new_map[k] = to_json(v)157                json_data = new_map158            super(Map, self).__init__(json_data, _readonly=_readonly)159        def clear(self):160            with self._lock:161                self._cached_wrappers.clear()162                self._json_data.clear()163        def keys(self):164            return six.viewkeys(self._json_data)165        def iteritems(self):166            for key in self:167                yield (key, self[key])168        def itervalues(self):169            for key in self:170                yield self[key]171        def get(self, key, default_value=None):172            with self._lock:173                if key in self._json_data:174                    return self[key]175                return default_value176        def __len__(self):177            return len(self._json_data)178        def __contains__(self, key):179            return key in self._json_data180        def __getitem__(self, key):181            with self._lock:182                if key not in self._json_data:183                    raise KeyError(key)184                return self._get_wrapped(key, wrapped_type)185        def __setitem__(self, key, value):186            with self._lock:187                self._set_wrapped(key, value, validator)188                self._json_data[key] = None # placeholder189        def __delitem__(self, key):190            if self._readonly:191                raise AttributeError192            with self._lock:193                del self._json_data[key]194                self._cached_wrappers.pop(key, None)195        def __iter__(self):196            return iter(self._json_data)197    return Map198def typed_map(key_type, value_type, key_validator=None, value_validator=None):199    key_validator = _normalize_validator(key_type, key_validator)200    value_validator = _normalize_validator(value_type, value_validator)201    class Map(JsonObjectWrapper, MapBase):202        supports_validation = True203        __slots__ = ()204        def __init__(self, json_data=None, _readonly=False):205            if isinstance(json_data, MapBase):206                json_data = json_data.to_json()207            elif json_data is not None:208                new_map = collections.OrderedDict()209                for k, v in six.viewitems(json_data):210                    key_validator(k)211                    value_validator(v)212                    new_map[str(k)] = to_json(v)213                json_data = new_map214            super(Map, self).__init__(json_data, _readonly=_readonly)215        def clear(self):216            with self._lock:217                self._cached_wrappers.clear()218                self._json_data.clear()219        def keys(self):220            return [key_validator(k) for k in six.viewkeys(self._json_data)]221        def iteritems(self):222            for key in self:223                yield (key, self[key])224        def itervalues(self):225            for key in self:226                yield self[key]227        def get(self, key, default_value=None):228            key = str(key)229            with self._lock:230                if key in self._json_data:231                    return self._get_wrapped(key, value_type)232                return default_value233        def __len__(self):234            return len(self._json_data)235        def __contains__(self, key):236            return str(key) in self._json_data237        def __getitem__(self, key):238            key = str(key)239            with self._lock:240                if key not in self._json_data:241                    raise KeyError(key)242                return self._get_wrapped(key, value_type)243        def __setitem__(self, key, value):244            key = str(key)245            with self._lock:246                self._set_wrapped(key, value, value_validator)247                self._json_data[key] = None # placeholder248        def __delitem__(self, key):249            if self._readonly:250                raise AttributeError251            key = str(key)252            with self._lock:253                del self._json_data[key]254                self._cached_wrappers.pop(key, None)255        def __iter__(self):256            for key in self._json_data:257                yield key_validator(key)258    return Map259def typed_set(wrapped_type):260    def wrapper(x, _readonly=False):261        set_type = frozenset if _readonly else set262        kwargs = dict()263        if hasattr(wrapped_type, 'supports_readonly'):264            kwargs.update(_readonly=True)265        if x is None:266            return set_type()267        return set_type(wrapped_type(v, **kwargs) for v in x)268    wrapper.supports_readonly = True269    return wrapper270def typed_list(wrapped_type, validator=None):271    validator = _normalize_validator(wrapped_type, validator)272    class TypedList(object):273        supports_readonly = True274        supports_validation = True275        __slots__ = ('_readonly', '_data')276        def __init__(self, json_data=None, _readonly=False):277            if json_data is None:278                json_data = []279            if not isinstance(json_data, (list, tuple, np.ndarray)):280                raise ValueError281            self._readonly = _readonly282            self._data = [validator(x) for x in json_data]283        def __len__(self):284            return len(self._data)285        def __getitem__(self, key):286            return self._data[key]287        def __delitem__(self, key):288            if self._readonly:289                raise AttributeError290            del self._data[key]291        def __setitem__(self, key, value):292            if self._readonly:293                raise AttributeError294            if isinstance(key, slice):295                values = [validator(x) for x in value]296                self._data[key] = values297            else:298                value = validator(x)299                self._data[key] = value300        def __iter__(self):301            return iter(self._data)302        def append(self, x):303            if self._readonly:304                raise AttributeError305            x = validator(x)306            self._data.append(x)307        def extend(self, values):308            for x in values:309                self.append(x)310        def insert(self, index, x):311            x = validator(x)312            self._data.insert(index, x)313        def pop(self, index=-1):314            return self._data.pop(index)315        def to_json(self):316            return [to_json(x) for x in self._data]317        def __deepcopy__(self, memo):318            return type(self)(copy.deepcopy(self.to_json(), memo))319        def __repr__(self):320            return encode_json_for_repr(self.to_json())321    return TypedList322def number_or_string(value):323    if (not isinstance(value, numbers.Real) and not isinstance(value, six.text_type)):324        raise TypeError...

read_only_dict.py

Source:read_only_dict.py

1"""Read only dictionary."""2from typing import Any, TypeVar3def _readonly(*args: Any, **kwargs: Any) -> Any:4    """Raise an exception when a read only dict is modified."""5    raise RuntimeError("Cannot modify ReadOnlyDict")6Key = TypeVar("Key")7Value = TypeVar("Value")8class ReadOnlyDict(dict[Key, Value]):9    """Read only version of dict that is compatible with dict types."""10    __setitem__ = _readonly11    __delitem__ = _readonly12    pop = _readonly13    popitem = _readonly14    clear = _readonly15    update = _readonly...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.