<s> import os <EOL> import logging <EOL> import json <EOL> from contextlib import contextmanager <EOL> import six <EOL> from pyramid . config import Configurator <EOL> log = logging . getLogger ( __name__ ) <EOL> def get_json_encoder ( ) : <EOL> try : <EOL> from nefertari import engine <EOL> return engine . JSONEncoder <EOL> except AttributeError : <EOL> from nefertari . renderers import _JSONEncoder <EOL> return _JSONEncoder <EOL> def json_dumps ( body , encoder = None ) : <EOL> if encoder is None : <EOL> encoder = get_json_encoder ( ) <EOL> return json . dumps ( body , cls = encoder ) <EOL> def split_strip ( _str , on = '<STR_LIT:U+002C>' ) : <EOL> lst = _str if isinstance ( _str , list ) else _str . split ( on ) <EOL> return list ( filter ( bool , [ e . strip ( ) for e in lst ] ) ) <EOL> def process_limit ( start , page , limit ) : <EOL> try : <EOL> limit = int ( limit ) <EOL> if start is not None and page is not None : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> if start is not None : <EOL> start = int ( start ) <EOL> elif page is not None : <EOL> start = int ( page ) * limit <EOL> else : <EOL> start = <NUM_LIT:0> <EOL> if limit < <NUM_LIT:0> or start < <NUM_LIT:0> : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> except ( ValueError , TypeError ) as e : <EOL> raise ValueError ( e ) <EOL> return start , limit <EOL> def extend_list ( param ) : <EOL> _new = [ ] <EOL> if isinstance ( param , ( list , set ) ) : <EOL> for each in param : <EOL> if isinstance ( each , six . string_types ) and each . find ( '<STR_LIT:U+002C>' ) != - <NUM_LIT:1> : <EOL> _new . extend ( split_strip ( each ) ) <EOL> else : <EOL> _new . append ( each ) <EOL> elif isinstance ( param , six . string_types ) and param . find ( '<STR_LIT:U+002C>' ) != - <NUM_LIT:1> : <EOL> _new = split_strip ( param ) <EOL> return _new <EOL> def process_fields ( _fields ) : <EOL> fields_only = [ ] <EOL> fields_exclude = [ ] <EOL> if isinstance ( _fields , six . string_types ) : <EOL> _fields = split_strip ( _fields ) <EOL> for field in extend_list ( _fields ) : <EOL> field = field . strip ( ) <EOL> if not field : <EOL> continue <EOL> if field [ <NUM_LIT:0> ] == "<STR_LIT:->" : <EOL> fields_exclude . append ( field [ <NUM_LIT:1> : ] ) <EOL> else : <EOL> fields_only . append ( field ) <EOL> return fields_only , fields_exclude <EOL> def snake2camel ( text ) : <EOL> "<STR_LIT>" <EOL> return '<STR_LIT>' . join ( [ a . title ( ) for a in text . split ( "<STR_LIT:_>" ) ] ) <EOL> def maybe_dotted ( module , throw = True ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return Configurator ( ) . maybe_dotted ( module ) <EOL> except ImportError as e : <EOL> err = '<STR_LIT>' % ( module , e ) <EOL> if throw : <EOL> raise ImportError ( err ) <EOL> else : <EOL> log . error ( err ) <EOL> return None <EOL> @ contextmanager <EOL> def chdir ( path ) : <EOL> old_dir = os . getcwd ( ) <EOL> os . chdir ( path ) <EOL> yield <EOL> os . chdir ( old_dir ) <EOL> def isnumeric ( value ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> float ( value ) <EOL> return True <EOL> except ( ValueError , TypeError ) : <EOL> return False <EOL> def issequence ( arg ) : <EOL> """<STR_LIT>""" <EOL> string_behaviour = ( <EOL> isinstance ( arg , six . string_types ) or <EOL> isinstance ( arg , six . text_type ) ) <EOL> list_behaviour = hasattr ( arg , '<STR_LIT>' ) or hasattr ( arg , '<STR_LIT>' ) <EOL> return not string_behaviour and list_behaviour <EOL> def merge_dicts ( a , b , path = None ) : <EOL> """<STR_LIT>""" <EOL> if path is None : <EOL> path = [ ] <EOL> for key in b : <EOL> if key in a : <EOL> if isinstance ( a [ key ] , dict ) and isinstance ( b [ key ] , dict ) : <EOL> merge_dicts ( a [ key ] , b [ key ] , path + [ str ( key ) ] ) <EOL> elif a [ key ] == b [ key ] : <EOL> pass <EOL> else : <EOL> raise Exception ( <EOL> '<STR_LIT>' % '<STR_LIT:.>' . join ( path + [ str ( key ) ] ) ) <EOL> else : <EOL> a [ key ] = b [ key ] <EOL> return a <EOL> def str2dict ( dotted_str , value = None , separator = '<STR_LIT:.>' ) : <EOL> """<STR_LIT>""" <EOL> dict_ = { } <EOL> parts = dotted_str . split ( separator ) <EOL> d , prev = dict_ , None <EOL> for part in parts : <EOL> prev = d <EOL> d = d . setdefault ( part , { } ) <EOL> else : <EOL> if value is not None : <EOL> prev [ part ] = value <EOL> return dict_ <EOL> def validate_data_privacy ( request , data , wrapper_kw = None ) : <EOL> """<STR_LIT>""" <EOL> from nefertari import wrappers <EOL> if wrapper_kw is None : <EOL> wrapper_kw = { } <EOL> wrapper = wrappers . apply_privacy ( request ) <EOL> allowed_fields = wrapper ( result = data , ** wrapper_kw ) . keys ( ) <EOL> data = data . copy ( ) <EOL> data . pop ( '<STR_LIT>' , None ) <EOL> not_allowed_fields = set ( data . keys ( ) ) - set ( allowed_fields ) <EOL> if not_allowed_fields : <EOL> raise wrappers . ValidationError ( '<STR_LIT:U+002CU+0020>' . join ( not_allowed_fields ) ) <EOL> def drop_reserved_params ( params ) : <EOL> """<STR_LIT>""" <EOL> from nefertari import RESERVED_PARAMS <EOL> params = params . copy ( ) <EOL> for reserved_param in RESERVED_PARAMS : <EOL> if reserved_param in params : <EOL> params . pop ( reserved_param ) <EOL> return params <EOL> def is_document ( data ) : <EOL> """<STR_LIT>""" <EOL> return isinstance ( data , dict ) and '<STR_LIT>' in data </s>
