<s> import sys <EOL> import threading <EOL> import warnings <EOL> from collections import Counter , OrderedDict , defaultdict <EOL> from functools import partial <EOL> from django . core . exceptions import AppRegistryNotReady , ImproperlyConfigured <EOL> from django . utils import lru_cache <EOL> from . config import AppConfig <EOL> class Apps ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , installed_apps = ( ) ) : <EOL> if installed_apps is None and hasattr ( sys . modules [ __name__ ] , '<STR_LIT>' ) : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> self . all_models = defaultdict ( OrderedDict ) <EOL> self . app_configs = OrderedDict ( ) <EOL> self . stored_app_configs = [ ] <EOL> self . apps_ready = self . models_ready = self . ready = False <EOL> self . _lock = threading . Lock ( ) <EOL> self . _pending_operations = defaultdict ( list ) <EOL> if installed_apps is not None : <EOL> self . populate ( installed_apps ) <EOL> def populate ( self , installed_apps = None ) : <EOL> """<STR_LIT>""" <EOL> if self . ready : <EOL> return <EOL> with self . _lock : <EOL> if self . ready : <EOL> return <EOL> if self . app_configs : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> for entry in installed_apps : <EOL> if isinstance ( entry , AppConfig ) : <EOL> app_config = entry <EOL> else : <EOL> app_config = AppConfig . create ( entry ) <EOL> if app_config . label in self . app_configs : <EOL> raise ImproperlyConfigured ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % app_config . label ) <EOL> self . app_configs [ app_config . label ] = app_config <EOL> counts = Counter ( <EOL> app_config . name for app_config in self . app_configs . values ( ) ) <EOL> duplicates = [ <EOL> name for name , count in counts . most_common ( ) if count > <NUM_LIT:1> ] <EOL> if duplicates : <EOL> raise ImproperlyConfigured ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % "<STR_LIT:U+002CU+0020>" . join ( duplicates ) ) <EOL> self . apps_ready = True <EOL> for app_config in self . app_configs . values ( ) : <EOL> all_models = self . all_models [ app_config . label ] <EOL> app_config . import_models ( all_models ) <EOL> self . clear_cache ( ) <EOL> self . models_ready = True <EOL> for app_config in self . get_app_configs ( ) : <EOL> app_config . ready ( ) <EOL> self . ready = True <EOL> def check_apps_ready ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . apps_ready : <EOL> raise AppRegistryNotReady ( "<STR_LIT>" ) <EOL> def check_models_ready ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . models_ready : <EOL> raise AppRegistryNotReady ( "<STR_LIT>" ) <EOL> def get_app_configs ( self ) : <EOL> """<STR_LIT>""" <EOL> self . check_apps_ready ( ) <EOL> return self . app_configs . values ( ) <EOL> def get_app_config ( self , app_label ) : <EOL> """<STR_LIT>""" <EOL> self . check_apps_ready ( ) <EOL> try : <EOL> return self . app_configs [ app_label ] <EOL> except KeyError : <EOL> message = "<STR_LIT>" % app_label <EOL> for app_config in self . get_app_configs ( ) : <EOL> if app_config . name == app_label : <EOL> message += "<STR_LIT>" % app_config . label <EOL> break <EOL> raise LookupError ( message ) <EOL> @ lru_cache . lru_cache ( maxsize = None ) <EOL> def get_models ( self , include_auto_created = False , <EOL> include_deferred = False , include_swapped = False ) : <EOL> """<STR_LIT>""" <EOL> self . check_models_ready ( ) <EOL> result = [ ] <EOL> for app_config in self . app_configs . values ( ) : <EOL> result . extend ( list ( app_config . get_models ( <EOL> include_auto_created , include_deferred , include_swapped ) ) ) <EOL> return result <EOL> def get_model ( self , app_label , model_name = None ) : <EOL> """<STR_LIT>""" <EOL> self . check_models_ready ( ) <EOL> if model_name is None : <EOL> app_label , model_name = app_label . split ( '<STR_LIT:.>' ) <EOL> return self . get_app_config ( app_label ) . get_model ( model_name . lower ( ) ) <EOL> def register_model ( self , app_label , model ) : <EOL> model_name = model . _meta . model_name <EOL> app_models = self . all_models [ app_label ] <EOL> if model_name in app_models : <EOL> if ( model . __name__ == app_models [ model_name ] . __name__ and <EOL> model . __module__ == app_models [ model_name ] . __module__ ) : <EOL> warnings . warn ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( app_label , model_name ) , <EOL> RuntimeWarning , stacklevel = <NUM_LIT:2> ) <EOL> else : <EOL> raise RuntimeError ( <EOL> "<STR_LIT>" % <EOL> ( model_name , app_label , app_models [ model_name ] , model ) ) <EOL> app_models [ model_name ] = model <EOL> self . do_pending_operations ( model ) <EOL> self . clear_cache ( ) <EOL> def is_installed ( self , app_name ) : <EOL> """<STR_LIT>""" <EOL> self . check_apps_ready ( ) <EOL> return any ( ac . name == app_name for ac in self . app_configs . values ( ) ) <EOL> def get_containing_app_config ( self , object_name ) : <EOL> """<STR_LIT>""" <EOL> self . check_apps_ready ( ) <EOL> candidates = [ ] <EOL> for app_config in self . app_configs . values ( ) : <EOL> if object_name . startswith ( app_config . name ) : <EOL> subpath = object_name [ len ( app_config . name ) : ] <EOL> if subpath == '<STR_LIT>' or subpath [ <NUM_LIT:0> ] == '<STR_LIT:.>' : <EOL> candidates . append ( app_config ) <EOL> if candidates : <EOL> return sorted ( candidates , key = lambda ac : - len ( ac . name ) ) [ <NUM_LIT:0> ] <EOL> def get_registered_model ( self , app_label , model_name ) : <EOL> """<STR_LIT>""" <EOL> model = self . all_models [ app_label ] . get ( model_name . lower ( ) ) <EOL> if model is None : <EOL> raise LookupError ( <EOL> "<STR_LIT>" % ( app_label , model_name ) ) <EOL> return model <EOL> @ lru_cache . lru_cache ( maxsize = None ) <EOL> def get_swappable_settings_name ( self , to_string ) : <EOL> """<STR_LIT>""" <EOL> for model in self . get_models ( include_swapped = True ) : <EOL> swapped = model . _meta . swapped <EOL> if swapped and swapped == to_string : <EOL> return model . _meta . swappable <EOL> if model . _meta . swappable and model . _meta . label == to_string : <EOL> return model . _meta . swappable <EOL> return None <EOL> def set_available_apps ( self , available ) : <EOL> """<STR_LIT>""" <EOL> available = set ( available ) <EOL> installed = set ( app_config . name for app_config in self . get_app_configs ( ) ) <EOL> if not available . issubset ( installed ) : <EOL> raise ValueError ( <EOL> "<STR_LIT>" <EOL> % "<STR_LIT:U+002CU+0020>" . join ( available - installed ) <EOL> ) <EOL> self . stored_app_configs . append ( self . app_configs ) <EOL> self . app_configs = OrderedDict ( <EOL> ( label , app_config ) <EOL> for label , app_config in self . app_configs . items ( ) <EOL> if app_config . name in available ) <EOL> self . clear_cache ( ) <EOL> def unset_available_apps ( self ) : <EOL> """<STR_LIT>""" <EOL> self . app_configs = self . stored_app_configs . pop ( ) <EOL> self . clear_cache ( ) <EOL> def set_installed_apps ( self , installed ) : <EOL> """<STR_LIT>""" <EOL> if not self . ready : <EOL> raise AppRegistryNotReady ( "<STR_LIT>" ) <EOL> self . stored_app_configs . append ( self . app_configs ) <EOL> self . app_configs = OrderedDict ( ) <EOL> self . apps_ready = self . models_ready = self . ready = False <EOL> self . clear_cache ( ) <EOL> self . populate ( installed ) <EOL> def unset_installed_apps ( self ) : <EOL> """<STR_LIT>""" <EOL> self . app_configs = self . stored_app_configs . pop ( ) <EOL> self . apps_ready = self . models_ready = self . ready = True <EOL> self . clear_cache ( ) <EOL> def clear_cache ( self ) : <EOL> """<STR_LIT>""" <EOL> self . get_models . cache_clear ( ) <EOL> if self . ready : <EOL> for app_config in self . app_configs . values ( ) : <EOL> for model in app_config . get_models ( include_auto_created = True ) : <EOL> model . _meta . _expire_cache ( ) <EOL> def lazy_model_operation ( self , function , * model_keys ) : <EOL> """<STR_LIT>""" <EOL> model_key , more_models = model_keys [ <NUM_LIT:0> ] , model_keys [ <NUM_LIT:1> : ] <EOL> if more_models : <EOL> supplied_fn = function <EOL> def function ( model ) : <EOL> next_function = partial ( supplied_fn , model ) <EOL> if getattr ( supplied_fn , '<STR_LIT>' , None ) : <EOL> next_function . field = supplied_fn . keywords . get ( '<STR_LIT>' ) <EOL> self . lazy_model_operation ( next_function , * more_models ) <EOL> try : <EOL> model_class = self . get_registered_model ( * model_key ) <EOL> except LookupError : <EOL> self . _pending_operations [ model_key ] . append ( function ) <EOL> else : <EOL> function ( model_class ) <EOL> def do_pending_operations ( self , model ) : <EOL> """<STR_LIT>""" <EOL> key = model . _meta . app_label , model . _meta . model_name <EOL> for function in self . _pending_operations . pop ( key , [ ] ) : <EOL> function ( model ) <EOL> apps = Apps ( installed_apps = None ) </s>
