<s> import warnings <EOL> from functools import partial <EOL> from couchdbkit import ResourceNotFound <EOL> from dimagi . ext . couchdbkit import * <EOL> import itertools <EOL> from corehq . apps . cachehq . mixins import CachedCouchDocumentMixin <EOL> from dimagi . utils . couch . database import iter_docs <EOL> from dimagi . utils . couch . migration import SyncSQLToCouchMixin , SyncCouchToSQLMixin <EOL> from dimagi . utils . decorators . memoized import memoized <EOL> from datetime import datetime <EOL> from django . db import models , transaction <EOL> import json_field <EOL> from casexml . apps . case . cleanup import close_case <EOL> from corehq . form_processor . interfaces . supply import SupplyInterface <EOL> from corehq . form_processor . exceptions import CaseNotFound <EOL> from corehq . apps . commtrack . const import COMMTRACK_USERNAME <EOL> from corehq . apps . domain . models import Domain <EOL> from corehq . apps . products . models import SQLProduct <EOL> from corehq . toggles import LOCATION_TYPE_STOCK_RATES <EOL> from mptt . models import MPTTModel , TreeForeignKey , TreeManager <EOL> LOCATION_REPORTING_PREFIX = '<STR_LIT>' <EOL> class LocationTypeManager ( models . Manager ) : <EOL> def full_hierarchy ( self , domain ) : <EOL> """<STR_LIT>""" <EOL> hierarchy = { } <EOL> def insert_loc_type ( loc_type ) : <EOL> """<STR_LIT>""" <EOL> if not loc_type . parent_type : <EOL> lt_hierarchy = hierarchy <EOL> else : <EOL> lt_hierarchy = insert_loc_type ( loc_type . parent_type ) <EOL> if loc_type . id not in lt_hierarchy : <EOL> lt_hierarchy [ loc_type . id ] = ( loc_type , { } ) <EOL> return lt_hierarchy [ loc_type . id ] [ <NUM_LIT:1> ] <EOL> for loc_type in self . filter ( domain = domain ) . all ( ) : <EOL> insert_loc_type ( loc_type ) <EOL> return hierarchy <EOL> def by_domain ( self , domain ) : <EOL> """<STR_LIT>""" <EOL> ordered_loc_types = [ ] <EOL> def step_through_graph ( hierarchy ) : <EOL> for _ , ( loc_type , children ) in hierarchy . items ( ) : <EOL> ordered_loc_types . append ( loc_type ) <EOL> step_through_graph ( children ) <EOL> step_through_graph ( self . full_hierarchy ( domain ) ) <EOL> return ordered_loc_types <EOL> StockLevelField = partial ( models . DecimalField , max_digits = <NUM_LIT:10> , decimal_places = <NUM_LIT:1> ) <EOL> class LocationType ( models . Model ) : <EOL> domain = models . CharField ( max_length = <NUM_LIT:255> , db_index = True ) <EOL> name = models . CharField ( max_length = <NUM_LIT:255> ) <EOL> code = models . SlugField ( db_index = False , null = True ) <EOL> parent_type = models . ForeignKey ( '<STR_LIT>' , null = True ) <EOL> administrative = models . BooleanField ( default = False ) <EOL> shares_cases = models . BooleanField ( default = False ) <EOL> view_descendants = models . BooleanField ( default = False ) <EOL> last_modified = models . DateTimeField ( auto_now = True ) <EOL> emergency_level = StockLevelField ( default = <NUM_LIT:0.5> ) <EOL> understock_threshold = StockLevelField ( default = <NUM_LIT> ) <EOL> overstock_threshold = StockLevelField ( default = <NUM_LIT> ) <EOL> objects = LocationTypeManager ( ) <EOL> class Meta : <EOL> app_label = '<STR_LIT>' <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( LocationType , self ) . __init__ ( * args , ** kwargs ) <EOL> self . _administrative_old = self . administrative <EOL> @ property <EOL> @ memoized <EOL> def commtrack_enabled ( self ) : <EOL> return Domain . get_by_name ( self . domain ) . commtrack_enabled <EOL> def _populate_stock_levels ( self ) : <EOL> from corehq . apps . commtrack . models import CommtrackConfig <EOL> ct_config = CommtrackConfig . for_domain ( self . domain ) <EOL> if ( <EOL> ( ct_config is None ) <EOL> or ( not self . commtrack_enabled ) <EOL> or LOCATION_TYPE_STOCK_RATES . enabled ( self . domain ) <EOL> ) : <EOL> return <EOL> config = ct_config . stock_levels_config <EOL> self . emergency_level = config . emergency_level <EOL> self . understock_threshold = config . understock_threshold <EOL> self . overstock_threshold = config . overstock_threshold <EOL> def save ( self , * args , ** kwargs ) : <EOL> from . tasks import sync_administrative_status <EOL> if not self . code : <EOL> from corehq . apps . commtrack . util import unicode_slug <EOL> self . code = unicode_slug ( self . name ) <EOL> if not self . commtrack_enabled : <EOL> self . administrative = True <EOL> self . _populate_stock_levels ( ) <EOL> is_not_first_save = self . pk is not None <EOL> saved = super ( LocationType , self ) . save ( * args , ** kwargs ) <EOL> if is_not_first_save and self . _administrative_old != self . administrative : <EOL> sync_administrative_status . delay ( self ) <EOL> self . _administrative_old = self . administrative <EOL> return saved <EOL> def __unicode__ ( self ) : <EOL> return self . name <EOL> def __repr__ ( self ) : <EOL> return u"<STR_LIT>" . format ( <EOL> self . domain , <EOL> self . name , <EOL> self . administrative , <EOL> ) . encode ( '<STR_LIT:utf-8>' ) <EOL> @ property <EOL> @ memoized <EOL> def can_have_children ( self ) : <EOL> return LocationType . objects . filter ( parent_type = self ) . exists ( ) <EOL> class LocationQueriesMixin ( object ) : <EOL> def location_ids ( self ) : <EOL> return self . values_list ( '<STR_LIT>' , flat = True ) <EOL> def couch_locations ( self , wrapped = True ) : <EOL> """<STR_LIT>""" <EOL> warnings . warn ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> DeprecationWarning , <EOL> ) <EOL> ids = self . location_ids ( ) <EOL> locations = iter_docs ( Location . get_db ( ) , ids ) <EOL> if wrapped : <EOL> return itertools . imap ( Location . wrap , locations ) <EOL> return locations <EOL> class LocationQuerySet ( LocationQueriesMixin , models . query . QuerySet ) : <EOL> pass <EOL> class LocationManager ( LocationQueriesMixin , TreeManager ) : <EOL> def _get_base_queryset ( self ) : <EOL> return LocationQuerySet ( self . model , using = self . _db ) <EOL> def get_queryset ( self ) : <EOL> return ( self . _get_base_queryset ( ) <EOL> . order_by ( self . tree_id_attr , self . left_attr ) ) <EOL> def get_from_user_input ( self , domain , user_input ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return self . get ( domain = domain , site_code = user_input ) <EOL> except self . model . DoesNotExist : <EOL> return self . get ( domain = domain , name__iexact = user_input ) <EOL> def filter_by_user_input ( self , domain , user_input ) : <EOL> """<STR_LIT>""" <EOL> return ( self . filter ( domain = domain ) <EOL> . filter ( models . Q ( name__icontains = user_input ) | <EOL> models . Q ( site_code__icontains = user_input ) ) ) <EOL> def filter_path_by_user_input ( self , domain , user_input ) : <EOL> """<STR_LIT>""" <EOL> direct_matches = self . filter_by_user_input ( domain , user_input ) <EOL> return self . get_queryset_descendants ( direct_matches , include_self = True ) <EOL> class OnlyUnarchivedLocationManager ( LocationManager ) : <EOL> def get_queryset ( self ) : <EOL> return ( super ( OnlyUnarchivedLocationManager , self ) . get_queryset ( ) <EOL> . filter ( is_archived = False ) ) <EOL> class SQLLocation ( SyncSQLToCouchMixin , MPTTModel ) : <EOL> domain = models . CharField ( max_length = <NUM_LIT:255> , db_index = True ) <EOL> name = models . CharField ( max_length = <NUM_LIT:100> , null = True ) <EOL> location_id = models . CharField ( max_length = <NUM_LIT:100> , db_index = True , unique = True ) <EOL> _migration_couch_id_name = "<STR_LIT>" <EOL> location_type = models . ForeignKey ( LocationType ) <EOL> site_code = models . CharField ( max_length = <NUM_LIT:255> ) <EOL> external_id = models . CharField ( max_length = <NUM_LIT:255> , null = True ) <EOL> metadata = json_field . JSONField ( default = { } ) <EOL> created_at = models . DateTimeField ( auto_now_add = True ) <EOL> last_modified = models . DateTimeField ( auto_now = True ) <EOL> is_archived = models . BooleanField ( default = False ) <EOL> latitude = models . DecimalField ( max_digits = <NUM_LIT:20> , decimal_places = <NUM_LIT:10> , null = True ) <EOL> longitude = models . DecimalField ( max_digits = <NUM_LIT:20> , decimal_places = <NUM_LIT:10> , null = True ) <EOL> parent = TreeForeignKey ( '<STR_LIT>' , null = True , blank = True , related_name = '<STR_LIT>' ) <EOL> _products = models . ManyToManyField ( SQLProduct , null = True ) <EOL> stocks_all_products = models . BooleanField ( default = True ) <EOL> supply_point_id = models . CharField ( max_length = <NUM_LIT:255> , db_index = True , unique = True , null = True ) <EOL> objects = LocationManager ( ) <EOL> active_objects = OnlyUnarchivedLocationManager ( ) <EOL> @ classmethod <EOL> def _migration_get_fields ( cls ) : <EOL> return [ "<STR_LIT>" , "<STR_LIT:name>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" ] <EOL> @ classmethod <EOL> def _migration_get_couch_model_class ( cls ) : <EOL> return Location <EOL> def _migration_do_sync ( self ) : <EOL> couch_obj = self . _migration_get_or_create_couch_object ( ) <EOL> couch_obj . _sql_location_type = self . location_type <EOL> couch_obj . latitude = float ( self . latitude ) if self . latitude else None <EOL> couch_obj . longitude = float ( self . longitude ) if self . longitude else None <EOL> self . _migration_sync_to_couch ( couch_obj ) <EOL> @ transaction . atomic ( ) <EOL> def save ( self , * args , ** kwargs ) : <EOL> from corehq . apps . commtrack . models import sync_supply_point <EOL> self . supply_point_id = sync_supply_point ( self ) <EOL> sync_to_couch = kwargs . pop ( '<STR_LIT>' , True ) <EOL> kwargs [ '<STR_LIT>' ] = False <EOL> super ( SQLLocation , self ) . save ( * args , ** kwargs ) <EOL> if sync_to_couch : <EOL> self . _migration_do_sync ( ) <EOL> @ property <EOL> def lineage ( self ) : <EOL> return list ( self . get_ancestors ( ascending = True ) . location_ids ( ) ) <EOL> @ property <EOL> def get_id ( self ) : <EOL> return self . location_id <EOL> @ property <EOL> def products ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . stocks_all_products : <EOL> return SQLProduct . by_domain ( self . domain ) <EOL> else : <EOL> return self . _products . all ( ) <EOL> @ products . setter <EOL> def products ( self , value ) : <EOL> self . stocks_all_products = ( set ( value ) == <EOL> set ( SQLProduct . by_domain ( self . domain ) ) ) <EOL> self . _products = value <EOL> class Meta : <EOL> app_label = '<STR_LIT>' <EOL> unique_together = ( '<STR_LIT>' , '<STR_LIT>' , ) <EOL> def __unicode__ ( self ) : <EOL> return u"<STR_LIT>" . format ( self . name , self . domain ) <EOL> def __repr__ ( self ) : <EOL> return u"<STR_LIT>" . format ( <EOL> self . domain , <EOL> self . name , <EOL> self . location_type . name , <EOL> ) . encode ( '<STR_LIT:utf-8>' ) <EOL> @ property <EOL> def display_name ( self ) : <EOL> return u"<STR_LIT>" . format ( self . name , self . location_type . name ) <EOL> def archived_descendants ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . get_descendants ( ) . filter ( is_archived = True ) <EOL> def child_locations ( self , include_archive_ancestors = False ) : <EOL> """<STR_LIT>""" <EOL> children = self . get_children ( ) <EOL> return _filter_for_archived ( children , include_archive_ancestors ) <EOL> @ classmethod <EOL> def root_locations ( cls , domain , include_archive_ancestors = False ) : <EOL> roots = cls . objects . root_nodes ( ) . filter ( domain = domain ) <EOL> return _filter_for_archived ( roots , include_archive_ancestors ) <EOL> def get_path_display ( self ) : <EOL> return '<STR_LIT:/>' . join ( self . get_ancestors ( include_self = True ) <EOL> . values_list ( '<STR_LIT:name>' , flat = True ) ) <EOL> def _make_group_object ( self , user_id , case_sharing ) : <EOL> from corehq . apps . groups . models import UnsavableGroup <EOL> g = UnsavableGroup ( ) <EOL> g . domain = self . domain <EOL> g . users = [ user_id ] if user_id else [ ] <EOL> g . last_modified = datetime . utcnow ( ) <EOL> if case_sharing : <EOL> g . name = self . get_path_display ( ) + '<STR_LIT>' <EOL> g . _id = self . location_id <EOL> g . case_sharing = True <EOL> g . reporting = False <EOL> else : <EOL> g . name = self . get_path_display ( ) <EOL> g . _id = LOCATION_REPORTING_PREFIX + self . location_id <EOL> g . case_sharing = False <EOL> g . reporting = True <EOL> g . metadata = { <EOL> '<STR_LIT>' : self . location_type . name , <EOL> '<STR_LIT>' : self . name , <EOL> } <EOL> for key , val in self . metadata . items ( ) : <EOL> g . metadata [ '<STR_LIT>' + key ] = val <EOL> return g <EOL> def get_case_sharing_groups ( self , for_user_id = None ) : <EOL> if self . location_type . shares_cases : <EOL> yield self . case_sharing_group_object ( for_user_id ) <EOL> if self . location_type . view_descendants : <EOL> for sql_loc in self . get_descendants ( ) . filter ( location_type__shares_cases = True , is_archived = False ) : <EOL> yield sql_loc . case_sharing_group_object ( for_user_id ) <EOL> def case_sharing_group_object ( self , user_id = None ) : <EOL> """<STR_LIT>""" <EOL> return self . _make_group_object ( <EOL> user_id , <EOL> case_sharing = True , <EOL> ) <EOL> def reporting_group_object ( self , user_id = None ) : <EOL> """<STR_LIT>""" <EOL> return self . _make_group_object ( <EOL> user_id , <EOL> case_sharing = False , <EOL> ) <EOL> @ property <EOL> @ memoized <EOL> def couch_location ( self ) : <EOL> return Location . get ( self . location_id ) <EOL> def is_direct_ancestor_of ( self , location ) : <EOL> return ( location . get_ancestors ( include_self = True ) <EOL> . filter ( pk = self . pk ) . exists ( ) ) <EOL> @ classmethod <EOL> def by_domain ( cls , domain ) : <EOL> return cls . objects . filter ( domain = domain ) <EOL> @ property <EOL> def path ( self ) : <EOL> return list ( reversed ( self . lineage ) ) <EOL> @ classmethod <EOL> def by_location_id ( cls , location_id ) : <EOL> try : <EOL> return cls . objects . get ( location_id = location_id ) <EOL> except cls . DoesNotExist : <EOL> return None <EOL> def linked_supply_point ( self ) : <EOL> if not self . supply_point_id : <EOL> return None <EOL> try : <EOL> return SupplyInterface ( self . domain ) . get_supply_point ( self . supply_point_id ) <EOL> except CaseNotFound : <EOL> return None <EOL> @ property <EOL> def location_type_object ( self ) : <EOL> return self . location_type <EOL> @ property <EOL> def location_type_name ( self ) : <EOL> return self . location_type . name <EOL> def _filter_for_archived ( locations , include_archive_ancestors ) : <EOL> """<STR_LIT>""" <EOL> if include_archive_ancestors : <EOL> return [ <EOL> item for item in locations <EOL> if item . is_archived or item . archived_descendants ( ) <EOL> ] <EOL> else : <EOL> return locations . filter ( is_archived = False ) <EOL> class Location ( SyncCouchToSQLMixin , CachedCouchDocumentMixin , Document ) : <EOL> domain = StringProperty ( ) <EOL> name = StringProperty ( ) <EOL> site_code = StringProperty ( ) <EOL> external_id = StringProperty ( ) <EOL> metadata = DictProperty ( ) <EOL> last_modified = DateTimeProperty ( ) <EOL> is_archived = BooleanProperty ( default = False ) <EOL> latitude = FloatProperty ( ) <EOL> longitude = FloatProperty ( ) <EOL> lineage = StringListProperty ( ) <EOL> previous_parents = StringListProperty ( ) <EOL> @ classmethod <EOL> def wrap ( cls , data ) : <EOL> last_modified = data . get ( '<STR_LIT>' ) <EOL> data . pop ( '<STR_LIT>' , None ) <EOL> from corehq . apps . groups . models import dt_no_Z_re <EOL> if last_modified and dt_no_Z_re . match ( last_modified ) : <EOL> data [ '<STR_LIT>' ] += '<STR_LIT>' <EOL> return super ( Location , cls ) . wrap ( data ) <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> from corehq . apps . locations . util import get_lineage_from_location , get_lineage_from_location_id <EOL> if '<STR_LIT>' in kwargs : <EOL> parent = kwargs [ '<STR_LIT>' ] <EOL> if parent : <EOL> if isinstance ( parent , Document ) : <EOL> lineage = get_lineage_from_location ( parent ) <EOL> else : <EOL> lineage = get_lineage_from_location_id ( parent ) <EOL> else : <EOL> lineage = [ ] <EOL> kwargs [ '<STR_LIT>' ] = lineage <EOL> del kwargs [ '<STR_LIT>' ] <EOL> location_type = kwargs . pop ( '<STR_LIT>' , None ) <EOL> super ( Document , self ) . __init__ ( * args , ** kwargs ) <EOL> if location_type : <EOL> self . location_type = location_type <EOL> def __unicode__ ( self ) : <EOL> return u"<STR_LIT>" . format ( self . name , self . domain ) <EOL> def __repr__ ( self ) : <EOL> return u"<STR_LIT>" . format ( <EOL> self . domain , <EOL> self . name , <EOL> self . location_type_name , <EOL> ) . encode ( '<STR_LIT:utf-8>' ) <EOL> def __eq__ ( self , other ) : <EOL> if isinstance ( other , Location ) : <EOL> return self . _id == other . _id <EOL> else : <EOL> return False <EOL> def __hash__ ( self ) : <EOL> return hash ( self . _id ) <EOL> @ property <EOL> def sql_location ( self ) : <EOL> return ( SQLLocation . objects . prefetch_related ( '<STR_LIT>' ) <EOL> . get ( location_id = self . _id ) ) <EOL> @ property <EOL> def location_id ( self ) : <EOL> return self . _id <EOL> @ property <EOL> def location_type ( self ) : <EOL> return self . location_type_object . name <EOL> _sql_location_type = None <EOL> @ location_type . setter <EOL> def location_type ( self , value ) : <EOL> msg = "<STR_LIT>" <EOL> if not value : <EOL> raise LocationType . DoesNotExist ( msg ) <EOL> try : <EOL> self . _sql_location_type = LocationType . objects . get ( <EOL> domain = self . domain , <EOL> name = value , <EOL> ) <EOL> except LocationType . DoesNotExist : <EOL> raise LocationType . DoesNotExist ( msg ) <EOL> def _archive_single_location ( self ) : <EOL> """<STR_LIT>""" <EOL> self . is_archived = True <EOL> self . save ( ) <EOL> self . _close_case_and_remove_users ( ) <EOL> def archive ( self ) : <EOL> """<STR_LIT>""" <EOL> for loc in [ self ] + self . descendants : <EOL> loc . _archive_single_location ( ) <EOL> def _unarchive_single_location ( self ) : <EOL> """<STR_LIT>""" <EOL> self . is_archived = False <EOL> self . save ( ) <EOL> sp = self . linked_supply_point ( ) <EOL> if sp and sp . closed : <EOL> for action in sp . actions : <EOL> if action . action_type == '<STR_LIT>' : <EOL> action . xform . archive ( user_id = COMMTRACK_USERNAME ) <EOL> break <EOL> def unarchive ( self ) : <EOL> """<STR_LIT>""" <EOL> for loc in [ self ] + self . descendants : <EOL> loc . _unarchive_single_location ( ) <EOL> def _close_case_and_remove_users ( self ) : <EOL> """<STR_LIT>""" <EOL> sp = self . linked_supply_point ( ) <EOL> if sp and not sp . closed : <EOL> close_case ( sp . case_id , self . domain , COMMTRACK_USERNAME ) <EOL> _unassign_users_from_location ( self . domain , self . _id ) <EOL> def full_delete ( self ) : <EOL> """<STR_LIT>""" <EOL> to_delete = [ self ] + self . descendants <EOL> with transaction . atomic ( ) : <EOL> for loc in to_delete : <EOL> loc . _close_case_and_remove_users ( ) <EOL> self . sql_location . get_descendants ( include_self = True ) . delete ( ) <EOL> Location . get_db ( ) . bulk_delete ( to_delete ) <EOL> @ classmethod <EOL> def _migration_get_fields ( cls ) : <EOL> return [ "<STR_LIT>" , "<STR_LIT:name>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> @ classmethod <EOL> def _migration_get_sql_model_class ( cls ) : <EOL> return SQLLocation <EOL> def _migration_do_sync ( self ) : <EOL> sql_location = self . _migration_get_or_create_sql_object ( ) <EOL> location_type = self . _sql_location_type or sql_location . location_type <EOL> sql_location . location_type = location_type <EOL> sql_location . parent = ( SQLLocation . objects . get ( location_id = self . parent_id ) <EOL> if self . parent_id else None ) <EOL> self . _migration_sync_to_sql ( sql_location ) <EOL> def save ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> self . last_modified = datetime . utcnow ( ) <EOL> if not self . site_code : <EOL> from corehq . apps . commtrack . util import generate_code <EOL> all_codes = [ <EOL> code . lower ( ) for code in <EOL> ( SQLLocation . objects . filter ( domain = self . domain ) <EOL> . values_list ( '<STR_LIT>' , flat = True ) ) <EOL> ] <EOL> self . site_code = generate_code ( self . name , all_codes ) <EOL> if not self . _id : <EOL> self . _id = self . get_db ( ) . server . next_uuid ( ) <EOL> sync_to_sql = kwargs . pop ( '<STR_LIT>' , True ) <EOL> kwargs [ '<STR_LIT>' ] = False <EOL> with transaction . atomic ( ) : <EOL> if sync_to_sql : <EOL> self . _migration_do_sync ( ) <EOL> super ( Location , self ) . save ( * args , ** kwargs ) <EOL> @ classmethod <EOL> def filter_by_type ( cls , domain , loc_type , root_loc = None ) : <EOL> if root_loc : <EOL> query = root_loc . sql_location . get_descendants ( include_self = True ) <EOL> else : <EOL> query = SQLLocation . objects <EOL> ids = ( query . filter ( domain = domain , location_type__name = loc_type ) <EOL> . location_ids ( ) ) <EOL> return ( <EOL> cls . wrap ( l ) for l in iter_docs ( cls . get_db ( ) , list ( ids ) ) <EOL> if not l . get ( '<STR_LIT>' , False ) <EOL> ) <EOL> @ classmethod <EOL> def by_domain ( cls , domain , include_docs = True ) : <EOL> relevant_ids = SQLLocation . objects . filter ( domain = domain ) . location_ids ( ) <EOL> if not include_docs : <EOL> return relevant_ids <EOL> else : <EOL> return ( <EOL> cls . wrap ( l ) for l in iter_docs ( cls . get_db ( ) , list ( relevant_ids ) ) <EOL> if not l . get ( '<STR_LIT>' , False ) <EOL> ) <EOL> @ classmethod <EOL> def by_site_code ( cls , domain , site_code ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return ( SQLLocation . objects . get ( domain = domain , <EOL> site_code__iexact = site_code ) <EOL> . couch_location ) <EOL> except SQLLocation . DoesNotExist : <EOL> return None <EOL> @ classmethod <EOL> def root_locations ( cls , domain ) : <EOL> """<STR_LIT>""" <EOL> return list ( SQLLocation . root_locations ( domain ) . couch_locations ( ) ) <EOL> @ classmethod <EOL> def get_in_domain ( cls , domain , id ) : <EOL> if id : <EOL> try : <EOL> loc = Location . get ( id ) <EOL> assert domain == loc . domain <EOL> return loc <EOL> except ( ResourceNotFound , AssertionError ) : <EOL> pass <EOL> return None <EOL> @ property <EOL> def is_root ( self ) : <EOL> return not self . lineage <EOL> @ property <EOL> def parent_id ( self ) : <EOL> if self . is_root : <EOL> return None <EOL> return self . lineage [ <NUM_LIT:0> ] <EOL> @ property <EOL> def parent ( self ) : <EOL> parent_id = self . parent_id <EOL> return Location . get ( parent_id ) if parent_id else None <EOL> def siblings ( self , parent = None ) : <EOL> if not parent : <EOL> parent = self . parent <EOL> locs = ( parent . children if parent else self . root_locations ( self . domain ) ) <EOL> return [ loc for loc in locs if loc . location_id != self . _id ] <EOL> @ property <EOL> def path ( self ) : <EOL> _path = list ( reversed ( self . lineage ) ) <EOL> _path . append ( self . _id ) <EOL> return _path <EOL> @ property <EOL> def descendants ( self ) : <EOL> """<STR_LIT>""" <EOL> return list ( self . sql_location . get_descendants ( ) . couch_locations ( ) ) <EOL> @ property <EOL> def children ( self ) : <EOL> """<STR_LIT>""" <EOL> return list ( SQLLocation . objects . filter ( parent = self . sql_location ) <EOL> . couch_locations ( ) ) <EOL> def linked_supply_point ( self ) : <EOL> return self . sql_location . linked_supply_point ( ) <EOL> @ property <EOL> def group_id ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _id <EOL> @ property <EOL> def location_type_object ( self ) : <EOL> return self . _sql_location_type or self . sql_location . location_type <EOL> @ property <EOL> def location_type_name ( self ) : <EOL> return self . location_type_object . name <EOL> def _unassign_users_from_location ( domain , location_id ) : <EOL> """<STR_LIT>""" <EOL> from corehq . apps . locations . dbaccessors import get_all_users_by_location <EOL> for user in get_all_users_by_location ( domain , location_id ) : <EOL> if user . is_web_user ( ) : <EOL> user . unset_location ( domain ) <EOL> elif user . is_commcare_user ( ) : <EOL> user . unset_location ( ) </s>
