<s> """<STR_LIT>""" <EOL> import inspect <EOL> import sys <EOL> import warnings <EOL> from functools import wraps <EOL> from django . db import models <EOL> from django . db . models . signals import class_prepared <EOL> from django . utils . functional import curry <EOL> from django_fsm . signals import pre_transition , post_transition <EOL> try : <EOL> from django . apps import apps as django_apps <EOL> def get_model ( app_label , model_name ) : <EOL> app = django_apps . get_app_config ( app_label ) <EOL> return app . get_model ( model_name ) <EOL> except ImportError : <EOL> from django . db . models . loading import get_model <EOL> __all__ = [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] <EOL> if sys . version_info [ : <NUM_LIT:2> ] == ( <NUM_LIT:2> , <NUM_LIT:6> ) : <EOL> def __getmembers ( object , predicate = None ) : <EOL> """<STR_LIT>""" <EOL> results = [ ] <EOL> for key in dir ( object ) : <EOL> try : <EOL> value = getattr ( object , key ) <EOL> except AttributeError : <EOL> continue <EOL> if not predicate or predicate ( value ) : <EOL> results . append ( ( key , value ) ) <EOL> results . sort ( ) <EOL> return results <EOL> inspect . getmembers = __getmembers <EOL> try : <EOL> from south . modelsinspector import add_introspection_rules <EOL> except ImportError : <EOL> pass <EOL> else : <EOL> add_introspection_rules ( [ ] , [ r"<STR_LIT>" ] ) <EOL> add_introspection_rules ( [ ] , [ r"<STR_LIT>" ] ) <EOL> add_introspection_rules ( [ ] , [ r"<STR_LIT>" ] ) <EOL> class TransitionNotAllowed ( Exception ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> self . object = kwargs . pop ( '<STR_LIT:object>' , None ) <EOL> self . method = kwargs . pop ( '<STR_LIT>' , None ) <EOL> super ( TransitionNotAllowed , self ) . __init__ ( * args , ** kwargs ) <EOL> class ConcurrentTransition ( Exception ) : <EOL> """<STR_LIT>""" <EOL> class Transition ( object ) : <EOL> def __init__ ( self , method , source , target , on_error , conditions , permission , custom ) : <EOL> self . method = method <EOL> self . source = source <EOL> self . target = target <EOL> self . on_error = on_error <EOL> self . conditions = conditions <EOL> self . permission = permission <EOL> self . custom = custom <EOL> @ property <EOL> def name ( self ) : <EOL> return self . method . __name__ <EOL> def has_perm ( self , instance , user ) : <EOL> if not self . permission : <EOL> return True <EOL> elif callable ( self . permission ) : <EOL> try : <EOL> return bool ( self . permission ( instance , user ) ) <EOL> except TypeError : <EOL> warnings . warn ( '<STR_LIT>' ) <EOL> return bool ( self . permission ( user ) ) <EOL> elif user . has_perm ( self . permission , instance ) : <EOL> return True <EOL> elif user . has_perm ( self . permission ) : <EOL> return True <EOL> else : <EOL> return False <EOL> def get_available_FIELD_transitions ( instance , field ) : <EOL> """<STR_LIT>""" <EOL> curr_state = field . get_state ( instance ) <EOL> transitions = field . transitions [ instance . __class__ ] <EOL> for name , transition in transitions . items ( ) : <EOL> meta = transition . _django_fsm <EOL> if meta . has_transition ( curr_state ) and meta . conditions_met ( instance , curr_state ) : <EOL> yield meta . get_transition ( curr_state ) <EOL> def get_all_FIELD_transitions ( instance , field ) : <EOL> """<STR_LIT>""" <EOL> return field . get_all_transitions ( instance . __class__ ) <EOL> def get_available_user_FIELD_transitions ( instance , user , field ) : <EOL> """<STR_LIT>""" <EOL> for transition in get_available_FIELD_transitions ( instance , field ) : <EOL> if transition . has_perm ( instance , user ) : <EOL> yield transition <EOL> class FSMMeta ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , field , method ) : <EOL> self . field = field <EOL> self . transitions = { } <EOL> def get_transition ( self , source ) : <EOL> transition = self . transitions . get ( source , None ) <EOL> if transition is None : <EOL> transition = self . transitions . get ( '<STR_LIT:*>' , None ) <EOL> if transition is None : <EOL> transition = self . transitions . get ( '<STR_LIT:+>' , None ) <EOL> return transition <EOL> def add_transition ( self , method , source , target , on_error = None , conditions = [ ] , permission = None , custom = { } ) : <EOL> if source in self . transitions : <EOL> raise AssertionError ( '<STR_LIT>' . format ( source ) ) <EOL> self . transitions [ source ] = Transition ( <EOL> method = method , <EOL> source = source , <EOL> target = target , <EOL> on_error = on_error , <EOL> conditions = conditions , <EOL> permission = permission , <EOL> custom = custom ) <EOL> def has_transition ( self , state ) : <EOL> """<STR_LIT>""" <EOL> if state in self . transitions : <EOL> return True <EOL> if '<STR_LIT:*>' in self . transitions : <EOL> return True <EOL> if '<STR_LIT:+>' in self . transitions and self . transitions [ '<STR_LIT:+>' ] . target != state : <EOL> return True <EOL> return False <EOL> def conditions_met ( self , instance , state ) : <EOL> """<STR_LIT>""" <EOL> transition = self . get_transition ( state ) <EOL> if transition is None : <EOL> return False <EOL> elif transition . conditions is None : <EOL> return True <EOL> else : <EOL> return all ( map ( lambda condition : condition ( instance ) , transition . conditions ) ) <EOL> def has_transition_perm ( self , instance , state , user ) : <EOL> transition = self . get_transition ( state ) <EOL> if not transition : <EOL> return False <EOL> else : <EOL> return transition . has_perm ( instance , user ) <EOL> def next_state ( self , current_state ) : <EOL> transition = self . get_transition ( current_state ) <EOL> if transition is None : <EOL> raise TransitionNotAllowed ( '<STR_LIT>' . format ( current_state ) ) <EOL> return transition . target <EOL> def exception_state ( self , current_state ) : <EOL> transition = self . get_transition ( current_state ) <EOL> if transition is None : <EOL> raise TransitionNotAllowed ( '<STR_LIT>' . format ( current_state ) ) <EOL> return transition . on_error <EOL> class FSMFieldDescriptor ( object ) : <EOL> def __init__ ( self , field ) : <EOL> self . field = field <EOL> def __get__ ( self , instance , type = None ) : <EOL> if instance is None : <EOL> raise AttributeError ( '<STR_LIT>' ) <EOL> return self . field . get_state ( instance ) <EOL> def __set__ ( self , instance , value ) : <EOL> if self . field . protected and self . field . name in instance . __dict__ : <EOL> raise AttributeError ( '<STR_LIT>' . format ( self . field . name ) ) <EOL> self . field . set_proxy ( instance , value ) <EOL> self . field . set_state ( instance , value ) <EOL> class FSMFieldMixin ( object ) : <EOL> descriptor_class = FSMFieldDescriptor <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> self . protected = kwargs . pop ( '<STR_LIT>' , False ) <EOL> self . transitions = { } <EOL> self . state_proxy = { } <EOL> state_choices = kwargs . pop ( '<STR_LIT>' , None ) <EOL> choices = kwargs . get ( '<STR_LIT>' , None ) <EOL> if state_choices is not None and choices is not None : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if state_choices is not None : <EOL> choices = [ ] <EOL> for state , title , proxy_cls_ref in state_choices : <EOL> choices . append ( ( state , title ) ) <EOL> self . state_proxy [ state ] = proxy_cls_ref <EOL> kwargs [ '<STR_LIT>' ] = choices <EOL> super ( FSMFieldMixin , self ) . __init__ ( * args , ** kwargs ) <EOL> def deconstruct ( self ) : <EOL> name , path , args , kwargs = super ( FSMFieldMixin , self ) . deconstruct ( ) <EOL> if self . protected : <EOL> kwargs [ '<STR_LIT>' ] = self . protected <EOL> return name , path , args , kwargs <EOL> def get_state ( self , instance ) : <EOL> return instance . __dict__ [ self . name ] <EOL> def set_state ( self , instance , state ) : <EOL> instance . __dict__ [ self . name ] = state <EOL> def set_proxy ( self , instance , state ) : <EOL> """<STR_LIT>""" <EOL> if state in self . state_proxy : <EOL> state_proxy = self . state_proxy [ state ] <EOL> try : <EOL> app_label , model_name = state_proxy . split ( "<STR_LIT:.>" ) <EOL> except ValueError : <EOL> app_label = instance . _meta . app_label <EOL> model_name = state_proxy <EOL> model = get_model ( app_label , model_name ) <EOL> if model is None : <EOL> raise ValueError ( '<STR_LIT>' . format ( state_proxy ) ) <EOL> instance . __class__ = model <EOL> def change_state ( self , instance , method , * args , ** kwargs ) : <EOL> meta = method . _django_fsm <EOL> method_name = method . __name__ <EOL> current_state = self . get_state ( instance ) <EOL> if not meta . has_transition ( current_state ) : <EOL> raise TransitionNotAllowed ( <EOL> "<STR_LIT>" . format ( current_state , method_name ) , <EOL> object = instance , method = method ) <EOL> if not meta . conditions_met ( instance , current_state ) : <EOL> raise TransitionNotAllowed ( <EOL> "<STR_LIT>" . format ( method_name ) , <EOL> object = instance , method = method ) <EOL> next_state = meta . next_state ( current_state ) <EOL> signal_kwargs = { <EOL> '<STR_LIT>' : instance . __class__ , <EOL> '<STR_LIT>' : instance , <EOL> '<STR_LIT:name>' : method_name , <EOL> '<STR_LIT:source>' : current_state , <EOL> '<STR_LIT:target>' : next_state <EOL> } <EOL> pre_transition . send ( ** signal_kwargs ) <EOL> try : <EOL> result = method ( instance , * args , ** kwargs ) <EOL> if next_state is not None : <EOL> self . set_proxy ( instance , next_state ) <EOL> self . set_state ( instance , next_state ) <EOL> except Exception as exc : <EOL> exception_state = meta . exception_state ( current_state ) <EOL> if exception_state : <EOL> self . set_proxy ( instance , exception_state ) <EOL> self . set_state ( instance , exception_state ) <EOL> signal_kwargs [ '<STR_LIT:target>' ] = exception_state <EOL> signal_kwargs [ '<STR_LIT>' ] = exc <EOL> post_transition . send ( ** signal_kwargs ) <EOL> raise <EOL> else : <EOL> post_transition . send ( ** signal_kwargs ) <EOL> return result <EOL> def get_all_transitions ( self , instance_cls ) : <EOL> """<STR_LIT>""" <EOL> transitions = self . transitions [ instance_cls ] <EOL> for name , transition in transitions . items ( ) : <EOL> meta = transition . _django_fsm <EOL> for transition in meta . transitions . values ( ) : <EOL> yield transition <EOL> def contribute_to_class ( self , cls , name , virtual_only = False ) : <EOL> self . base_cls = cls <EOL> super ( FSMFieldMixin , self ) . contribute_to_class ( cls , name , virtual_only = virtual_only ) <EOL> setattr ( cls , self . name , self . descriptor_class ( self ) ) <EOL> setattr ( cls , '<STR_LIT>' . format ( self . name ) , <EOL> curry ( get_all_FIELD_transitions , field = self ) ) <EOL> setattr ( cls , '<STR_LIT>' . format ( self . name ) , <EOL> curry ( get_available_FIELD_transitions , field = self ) ) <EOL> setattr ( cls , '<STR_LIT>' . format ( self . name ) , <EOL> curry ( get_available_user_FIELD_transitions , field = self ) ) <EOL> class_prepared . connect ( self . _collect_transitions ) <EOL> def _collect_transitions ( self , * args , ** kwargs ) : <EOL> sender = kwargs [ '<STR_LIT>' ] <EOL> if not issubclass ( sender , self . base_cls ) : <EOL> return <EOL> def is_field_transition_method ( attr ) : <EOL> return ( inspect . ismethod ( attr ) or inspect . isfunction ( attr ) ) and hasattr ( attr , '<STR_LIT>' ) and attr . _django_fsm . field in [ self , self . name ] <EOL> sender_transitions = { } <EOL> transitions = inspect . getmembers ( sender , predicate = is_field_transition_method ) <EOL> for method_name , method in transitions : <EOL> method . _django_fsm . field = self <EOL> sender_transitions [ method_name ] = method <EOL> self . transitions [ sender ] = sender_transitions <EOL> class FSMField ( FSMFieldMixin , models . CharField ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> kwargs . setdefault ( '<STR_LIT:max_length>' , <NUM_LIT:50> ) <EOL> super ( FSMField , self ) . __init__ ( * args , ** kwargs ) <EOL> class FSMIntegerField ( FSMFieldMixin , models . IntegerField ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class FSMKeyField ( FSMFieldMixin , models . ForeignKey ) : <EOL> """<STR_LIT>""" <EOL> def get_state ( self , instance ) : <EOL> return instance . __dict__ [ self . attname ] <EOL> def set_state ( self , instance , state ) : <EOL> instance . __dict__ [ self . attname ] = self . to_python ( state ) <EOL> class ConcurrentTransitionMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( ConcurrentTransitionMixin , self ) . __init__ ( * args , ** kwargs ) <EOL> self . _update_initial_state ( ) <EOL> @ property <EOL> def state_fields ( self ) : <EOL> return filter ( <EOL> lambda field : isinstance ( field , FSMFieldMixin ) , <EOL> self . _meta . fields <EOL> ) <EOL> def _do_update ( self , base_qs , using , pk_val , values , update_fields , forced_update ) : <EOL> filter_on = filter ( lambda field : field . model == base_qs . model , self . state_fields ) <EOL> state_filter = dict ( ( field . attname , self . __initial_states [ field . attname ] ) for field in filter_on ) <EOL> updated = super ( ConcurrentTransitionMixin , self ) . _do_update ( <EOL> base_qs = base_qs . filter ( ** state_filter ) , <EOL> using = using , <EOL> pk_val = pk_val , <EOL> values = values , <EOL> update_fields = update_fields , <EOL> forced_update = forced_update <EOL> ) <EOL> if not updated and base_qs . filter ( pk = pk_val ) . exists ( ) : <EOL> raise ConcurrentTransition ( "<STR_LIT>" ) <EOL> return updated <EOL> def _update_initial_state ( self ) : <EOL> self . __initial_states = dict ( <EOL> ( field . attname , field . value_from_object ( self ) ) for field in self . state_fields <EOL> ) <EOL> def save ( self , * args , ** kwargs ) : <EOL> super ( ConcurrentTransitionMixin , self ) . save ( * args , ** kwargs ) <EOL> self . _update_initial_state ( ) <EOL> def transition ( field , source = '<STR_LIT:*>' , target = None , on_error = None , conditions = [ ] , permission = None , custom = { } ) : <EOL> """<STR_LIT>""" <EOL> def inner_transition ( func ) : <EOL> wrapper_installed , fsm_meta = True , getattr ( func , '<STR_LIT>' , None ) <EOL> if not fsm_meta : <EOL> wrapper_installed = False <EOL> fsm_meta = FSMMeta ( field = field , method = func ) <EOL> setattr ( func , '<STR_LIT>' , fsm_meta ) <EOL> if isinstance ( source , ( list , tuple ) ) : <EOL> for state in source : <EOL> func . _django_fsm . add_transition ( func , state , target , on_error , conditions , permission , custom ) <EOL> else : <EOL> func . _django_fsm . add_transition ( func , source , target , on_error , conditions , permission , custom ) <EOL> @ wraps ( func ) <EOL> def _change_state ( instance , * args , ** kwargs ) : <EOL> return fsm_meta . field . change_state ( instance , func , * args , ** kwargs ) <EOL> if not wrapper_installed : <EOL> return _change_state <EOL> return func <EOL> return inner_transition <EOL> def can_proceed ( bound_method , check_conditions = True ) : <EOL> """<STR_LIT>""" <EOL> if not hasattr ( bound_method , '<STR_LIT>' ) : <EOL> raise TypeError ( '<STR_LIT>' % bound_method . im_func . __name__ ) <EOL> meta = bound_method . _django_fsm <EOL> im_self = getattr ( bound_method , '<STR_LIT>' , getattr ( bound_method , '<STR_LIT>' ) ) <EOL> current_state = meta . field . get_state ( im_self ) <EOL> return meta . has_transition ( current_state ) and ( <EOL> not check_conditions or meta . conditions_met ( im_self , current_state ) ) <EOL> def has_transition_perm ( bound_method , user ) : <EOL> """<STR_LIT>""" <EOL> if not hasattr ( bound_method , '<STR_LIT>' ) : <EOL> raise TypeError ( '<STR_LIT>' % bound_method . im_func . __name__ ) <EOL> meta = bound_method . _django_fsm <EOL> im_self = getattr ( bound_method , '<STR_LIT>' , getattr ( bound_method , '<STR_LIT>' ) ) <EOL> current_state = meta . field . get_state ( im_self ) <EOL> return ( meta . has_transition ( current_state ) <EOL> and meta . conditions_met ( im_self , current_state ) <EOL> and meta . has_transition_perm ( im_self , current_state , user ) ) </s>
