Entity.has_changed() - Code Metrics - Inspection of "Modified refresh to search for performance..." - GEANT/met - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 7e5f4d...79309a )

by Andrea

created 2017-05-22 09:41 UTC

Entity.has_changed() B

↳ Parent: Entity

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
c	2
b	0
f	0
dl	0
loc	13
rs	8
cc	6

#################################################################

#
# This Software is Open Source. See License: https://github.com/TERENA/met/blob/master/LICENSE.md
# Copyright (c) 2012, TERENA All rights reserved.
#
# This Software is based on MET v1 developed for TERENA by Yaco Sistemas, http://www.yaco.es/
# MET v2 was developed for TERENA by Tamim Ziai, DAASI International GmbH, http://www.daasi.de
# Current version of MET has been revised for performance improvements by Andrea Biancini,
# Consortium GARR, http://www.garr.it
#########################################################################################

import simplejson as json

from urlparse import urlparse
from urllib import quote_plus
from datetime import datetime

from django.conf import settings
from django.core.cache import cache
from django.core.urlresolvers import reverse
from django.db import models
from django.db.models import Count
from django.db.models.query import QuerySet
from django.utils.translation import ugettext_lazy as _

from met.metadataparser.templatetags import attributemap

from base import JSONField, Base
from entity_type import EntityType
from entity_category import EntityCategory

TOP_LENGTH = getattr(settings, "TOP_LENGTH", 5)

def update_obj(mobj, obj, attrs=None):
    for_attrs = attrs or getattr(mobj, 'all_attrs', [])
    for attrb in attrs or for_attrs:
        if (getattr(mobj, attrb, None) and
            getattr(obj, attrb, None) and
            getattr(mobj, attrb) != getattr(obj, attrb)):
            setattr(obj, attrb, getattr(mobj, attrb))

class EntityQuerySet(QuerySet):
    def iterator(self):
        cached_federations = {}
        for entity in super(EntityQuerySet, self).iterator():
            if entity.file:
                continue

            federations = entity.federations.all()
            if federations:
                federation = federations[0]
            else:
                raise ValueError("Can't find entity metadata")

            for federation in federations:
                if not federation.id in cached_federations:
                    cached_federations[federation.id] = federation

                cached_federation = cached_federations[federation.id]
                try:
                    entity.load_metadata(federation=cached_federation)
                except ValueError:
                    # Allow entity in federation but not in federation file
                    continue
                else:
                    break

            yield entity

class EntityManager(models.Manager):
    def get_queryset(self):
        return EntityQuerySet(self.model, using=self._db)

class Entity(Base):
    READABLE_PROTOCOLS = {
        'urn:oasis:names:tc:SAML:1.1:protocol': 'SAML 1.1',
        'urn:oasis:names:tc:SAML:2.0:protocol': 'SAML 2.0',
        'urn:mace:shibboleth:1.0': 'Shiboleth 1.0',
    }

    entityid = models.CharField(blank=False, max_length=200, unique=True,
                                verbose_name=_(u'EntityID'), db_index=True)

    federations = models.ManyToManyField('Federation', through='Entity_Federations',
                                         verbose_name=_(u'Federations'))

    types = models.ManyToManyField('EntityType', verbose_name=_(u'Type'))

    name = JSONField(blank=True, null=True, max_length=2000,
                     verbose_name=_(u'Display Name'))

    certstats = models.CharField(blank=True, null=True, max_length=200,
                                 unique=False, verbose_name=_(u'Certificate Stats'))

    entity_categories = models.ManyToManyField('EntityCategory',
                                               verbose_name=_(u'Entity categories'))

    _display_protocols = models.CharField(blank=True, null=True, max_length=300,
                                          unique=False, verbose_name=_(u'Display Protocols'))

    objects = models.Manager()

    longlist = EntityManager()

    curfed = None

    @property
    def certificates(self):
        return json.loads(self.certstats)

    @property
    def registration_authority_xml(self):
        return self._get_property('registration_authority')

    @property
    def registration_policy(self):
        return self._get_property('registration_policy')

    @property
    def registration_instant(self):
        reginstant = self._get_property('registration_instant')
        if reginstant is None:
            return None
        reginstant = "%sZ" % reginstant[0:19]
        return datetime.strptime(reginstant, '%Y-%m-%dT%H:%M:%SZ')

    @property
    def protocols(self):
        try:
            return ' '.join(self._get_property('protocols'))
        except Exception:
            return ''

    @property
    def languages(self):
        try:
            return ' '.join(self._get_property('languages'))
        except Exception:
            return ''

    @property
    def scopes(self):
        try:
            return ' '.join(self._get_property('scopes'))
        except Exception:
            return ''

    @property
    def attributes(self):
        try:
            attributes = self._get_property('attr_requested')
            if not attributes:
                return []
            return attributes['required']
        except Exception:
            return []

    @property
    def attributes_optional(self):
        try:
            attributes = self._get_property('attr_requested')
            if not attributes:
                return []
            return attributes['optional']
        except Exception:
            return []

    @property
    def organization(self):
        organization = self._get_property('organization')
        if not organization:
            return []

        vals = []
        for lang, data in organization.items():
            data['lang'] = lang
            vals.append(data)

        return vals

    @property
    def display_name(self):
        try:
            return self._get_property('displayName')
        except Exception:
            return ''

    @property
    def federations_count(self):
        try:
            return str(self.federations.all().count())
        except Exception:
            return ''
        
    @property
    def description(self):
        try:
            return self._get_property('description')
        except Exception:
            return ''

    @property
    def info_url(self):
        try:
            return self._get_property('infoUrl')
        except Exception:
            return ''

    @property
    def privacy_url(self):
        try:
            return self._get_property('privacyUrl')
        except Exception:
            return ''

    @property
    def xml(self):
        try:
            return self._get_property('xml')
        except Exception:
            return ''

    @property
    def xml_types(self):
        try:
            return self._get_property('entity_types')
        except Exception:
            return []

    @property
    def xml_categories(self):
        try:
            return self._get_property('entity_categories')
        except Exception:
            return []

    @property
    def display_protocols(self):
        protocols = []

        #xml_protocols = self._get_property('protocols')
        xml_protocols = self._display_protocols
        if xml_protocols:
            for proto in xml_protocols.split(' '):
                protocols.append(self.READABLE_PROTOCOLS.get(proto, proto))

        return protocols

    def display_attributes(self):
        attributes = {}
        for [attr, friendly] in self.attributes:
            if friendly:
                attributes[attr] = friendly
            elif attr in attributemap.MAP['fro']:
                attributes[attr] = attributemap.MAP['fro'][attr]
            else:
                attributes[attr] = '?'
        return attributes

    def display_attributes_optional(self):
        attributes = {}
        for [attr, friendly] in self.attributes_optional:
            if friendly:
                attributes[attr] = friendly
            elif attr in attributemap.MAP['fro']:
                attributes[attr] = attributemap.MAP['fro'][attr]
            else:
                attributes[attr] = '?'
        return attributes

    @property
    def contacts(self):
        contacts = []
        for cur_contact in self._get_property('contacts'):
            if cur_contact['name'] and cur_contact['surname']:
                contact_name = '%s %s' % (cur_contact['name'], cur_contact['surname'])
            elif cur_contact['name']:
                contact_name = cur_contact['name']
            elif cur_contact['surname']:
                contact_name = cur_contact['surname']
            else:
                contact_name = urlparse(cur_contact['email']).path.partition('?')[0]
            c_type = 'undefined'
            if cur_contact['type']:
                c_type = cur_contact['type']
            contacts.append({ 'name': contact_name, 'email': cur_contact['email'], 'type': c_type })
        return contacts

    @property
    def logos(self):
        logos = []
        for cur_logo in self._get_property('logos'):
            cur_logo['external'] = True
            logos.append(cur_logo)

        return logos

    class Meta(object):
        verbose_name = _(u'Entity')
        verbose_name_plural = _(u'Entities')

    def __unicode__(self):
        return self.entityid

    def load_metadata(self, federation=None, entity_data=None):
        if hasattr(self, '_entity_cached'):
            return

        if self.file:
            self._entity_cached = self.load_file().get_entity(self.entityid)
        elif federation:
            self._entity_cached = federation.get_entity_metadata(self.entityid)
        elif entity_data:
            self._entity_cached = entity_data
        else:
            right_fed = None
            first_fed = None
            for fed in self.federations.all():
                if fed.registration_authority == self.registration_authority:
                    right_fed = fed
                if first_fed is None:
                    first_fed = fed

            if right_fed is not None:
                entity_cached = right_fed.get_entity_metadata(self.entityid)
                self._entity_cached = entity_cached
            else:
                entity_cached = first_fed.get_entity_metadata(self.entityid)
                self._entity_cached = entity_cached

        if not hasattr(self, '_entity_cached'):
            raise ValueError("Can't find entity metadata")

    def _get_property(self, prop, federation=None):
        try:
            self.load_metadata(federation or self.curfed)
        except ValueError:
            return None

        if hasattr(self, '_entity_cached'):
            return self._entity_cached.get(prop, None)
        else:
            raise ValueError("Not metadata loaded")

    def _get_or_create_etypes(self, cached_entity_types):
        entity_types = []
        cur_cached_types = [t.xmlname for t in self.types.all()]
        for etype in self.xml_types:
            if etype in cur_cached_types:
               break

            if cached_entity_types is None:
                entity_type, _ = EntityType.objects.get_or_create(xmlname=etype,
                                                                  name=DESCRIPTOR_TYPES_DISPLAY[etype])
            else:
                if etype in cached_entity_types:
                    entity_type = cached_entity_types[etype]
                else:
                    entity_type = EntityType.objects.create(xmlname=etype,
                                                            name=DESCRIPTOR_TYPES_DISPLAY[etype])
            entity_types.append(entity_type)
        return entity_types

    def _get_or_create_ecategories(self, cached_entity_categories):
        entity_categories = []
        cur_cached_categories = [t.category_id for t in self.entity_categories.all()]
        for ecategory in self.xml_categories:
            if ecategory in cur_cached_categories:
                break

            if cached_entity_categories is None:
                entity_category, _ = EntityCategory.objects.get_or_create(category_id=ecategory)
            else:
                if ecategory in cached_entity_categories:
                    entity_category = cached_entity_categories[ecategory]
                else:
                    entity_category = EntityCategory.objects.create(category_id=ecategory)
            entity_categories.append(entity_category)
        return entity_categories

    def process_metadata(self, auto_save=True, entity_data=None, cached_entity_types=None):
        if not entity_data:
            self.load_metadata()

        if self.entityid.lower() != entity_data.get('entityid').lower():
            raise ValueError("EntityID is not the same: %s != %s" % (self.entityid.lower(), entity_data.get('entityid').lower()))

        self._entity_cached = entity_data

        if self.xml_types:
            entity_types = self._get_or_create_etypes(cached_entity_types)
            if len(entity_types) > 0:
                self.types.add(*entity_types)

        if self.xml_categories:
            db_entity_categories = EntityCategory.objects.all()
            cached_entity_categories = { entity_category.category_id: entity_category for entity_category in db_entity_categories }

            # Delete categories no more present in XML
            self.entity_categories.clear()

            # Create all entities, if not alread in database
            entity_categories = self._get_or_create_ecategories(cached_entity_categories)

            # Add categories to entity
            if len(entity_categories) > 0:
                self.entity_categories.add(*entity_categories)
        else:
            # No categories in XML, delete eventual categorie sin DB
            self.entity_categories.clear()

        newname = self._get_property('displayName')
        if newname and newname != '':
            self.name = newname

        newprotocols = self.protocols
        if newprotocols and newprotocols != "":
            self._display_protocols = newprotocols

        self.certstats = self._get_property('certstats')

        if str(self._get_property('registration_authority')) != '':
            self.registration_authority = self._get_property('registration_authority')

        if auto_save:
            self.save()

    def to_dict(self):
        self.load_metadata()

        entity = self._entity_cached.copy()
        entity["types"] = [unicode(f) for f in self.types.all()]
        entity["federations"] = [{u"name": unicode(f), u"url": f.get_absolute_url()}
                                  for f in self.federations.all()]

        if self.registration_authority:
            entity["registration_authority"] = self.registration_authority
        if self.registration_instant:
            entity["registration_instant"] = '%s' % self.registration_instant

        if "file_id" in entity.keys():
            del entity["file_id"]
        if "entity_types" in entity.keys():
            del entity["entity_types"]

        return entity

    def display_etype(value, separator=', '):
            return separator.join([unicode(item) for item in value.all()])

    @classmethod
    def get_most_federated_entities(self, maxlength=TOP_LENGTH, cache_expire=None):
        entities = None
        if cache_expire:
            entities = cache.get("most_federated_entities")

        if not entities or len(entities) < maxlength:
            # Entities with count how many federations belongs to, and sorted by most first
            ob_entities = Entity.objects.all().annotate(federationslength=Count("federations")).order_by("-federationslength")
            ob_entities = ob_entities.prefetch_related('types', 'federations')
            ob_entities = ob_entities[:maxlength]

            entities = []
            for entity in ob_entities:
                entities.append({
                    'entityid': entity.entityid,
                    'name': entity.name,
                    'absolute_url': entity.get_absolute_url(),
                    'types': [unicode(item) for item in entity.types.all()],
                    'federations': [(unicode(item.name), item.get_absolute_url()) for item in entity.federations.all()],
                })

        if cache_expire:
            cache.set("most_federated_entities", entities, cache_expire)

        return entities[:maxlength]

    def get_absolute_url(self):
        return reverse('entity_view', args=[quote_plus(self.entityid.encode('utf-8'))])

    def can_edit(self, user, delete):
        permission = 'delete_entity' if delete else 'change_entity'
        if user.is_superuser or (user.has_perm('metadataparser.%s' % permission) and user in self.editor_users.all()):
            return True

        for federation in self.federations.all():
            if federation.can_edit(user, False):
                return True

        return False

    def has_changed(self, entityid, name, registration_authority, certstats, display_protocols):
        if self.entityid != entityid:
            return True
        if self.name != name:
            return True
        if self.registration_authority != registration_authority:
            return True
        if self.certstats != certstats:
            return True
        if self._display_protocols != display_protocols:
            return True

        return False


1			#################################################################
2
3			#
4			# This Software is Open Source. See License: https://github.com/TERENA/met/blob/master/LICENSE.md
5			# Copyright (c) 2012, TERENA All rights reserved.
6			#
7			# This Software is based on MET v1 developed for TERENA by Yaco Sistemas, http://www.yaco.es/
8			# MET v2 was developed for TERENA by Tamim Ziai, DAASI International GmbH, http://www.daasi.de
9			# Current version of MET has been revised for performance improvements by Andrea Biancini,
10			# Consortium GARR, http://www.garr.it
11			#########################################################################################
12
13			import simplejson as json
14
15			from urlparse import urlparse
16			from urllib import quote_plus
17			from datetime import datetime
18
19			from django.conf import settings
20			from django.core.cache import cache
21			from django.core.urlresolvers import reverse
22			from django.db import models
23			from django.db.models import Count
24			from django.db.models.query import QuerySet
25			from django.utils.translation import ugettext_lazy as _
26
27			from met.metadataparser.templatetags import attributemap
28
29			from base import JSONField, Base
30			from entity_type import EntityType
31			from entity_category import EntityCategory
32
33			TOP_LENGTH = getattr(settings, "TOP_LENGTH", 5)
34
35			def update_obj(mobj, obj, attrs=None):
36			for_attrs = attrs or getattr(mobj, 'all_attrs', [])
37			for attrb in attrs or for_attrs:
38			if (getattr(mobj, attrb, None) and
39			getattr(obj, attrb, None) and
40			getattr(mobj, attrb) != getattr(obj, attrb)):
41			setattr(obj, attrb, getattr(mobj, attrb))
42
43			class EntityQuerySet(QuerySet):
44			def iterator(self):
45			cached_federations = {}
46			for entity in super(EntityQuerySet, self).iterator():
47			if entity.file:
48			continue
49
50			federations = entity.federations.all()
51			if federations:
52			federation = federations[0]
53			else:
54			raise ValueError("Can't find entity metadata")
55
56			for federation in federations:
57			if not federation.id in cached_federations:
58			cached_federations[federation.id] = federation
59
60			cached_federation = cached_federations[federation.id]
61			try:
62			entity.load_metadata(federation=cached_federation)
63			except ValueError:
64			# Allow entity in federation but not in federation file
65			continue
66			else:
67			break
68
69			yield entity
70
71			class EntityManager(models.Manager):
72			def get_queryset(self):
73			return EntityQuerySet(self.model, using=self._db)
74
75			class Entity(Base):
76			READABLE_PROTOCOLS = {
77			'urn:oasis:names:tc:SAML:1.1:protocol': 'SAML 1.1',
78			'urn:oasis:names:tc:SAML:2.0:protocol': 'SAML 2.0',
79			'urn:mace:shibboleth:1.0': 'Shiboleth 1.0',
80			}
81
82			entityid = models.CharField(blank=False, max_length=200, unique=True,
83			verbose_name=_(u'EntityID'), db_index=True)
84
85			federations = models.ManyToManyField('Federation', through='Entity_Federations',
86			verbose_name=_(u'Federations'))
87
88			types = models.ManyToManyField('EntityType', verbose_name=_(u'Type'))
89
90			name = JSONField(blank=True, null=True, max_length=2000,
91			verbose_name=_(u'Display Name'))
92
93			certstats = models.CharField(blank=True, null=True, max_length=200,
94			unique=False, verbose_name=_(u'Certificate Stats'))
95
96			entity_categories = models.ManyToManyField('EntityCategory',
97			verbose_name=_(u'Entity categories'))
98
99			_display_protocols = models.CharField(blank=True, null=True, max_length=300,
100			unique=False, verbose_name=_(u'Display Protocols'))
101
102			objects = models.Manager()
103
104			longlist = EntityManager()
105
106			curfed = None
107
108			@property
109			def certificates(self):
110			return json.loads(self.certstats)
111
112			@property
113			def registration_authority_xml(self):
114			return self._get_property('registration_authority')
115
116			@property
117			def registration_policy(self):
118			return self._get_property('registration_policy')
119
120			@property
121			def registration_instant(self):
122			reginstant = self._get_property('registration_instant')
123			if reginstant is None:
124			return None
125			reginstant = "%sZ" % reginstant[0:19]
126			return datetime.strptime(reginstant, '%Y-%m-%dT%H:%M:%SZ')
127
128			@property
129			def protocols(self):
130			try:
131			return ' '.join(self._get_property('protocols'))
132			except Exception:
133			return ''
134
135			@property
136			def languages(self):
137			try:
138			return ' '.join(self._get_property('languages'))
139			except Exception:
140			return ''
141
142			@property
143			def scopes(self):
144			try:
145			return ' '.join(self._get_property('scopes'))
146			except Exception:
147			return ''
148
149			@property
150			def attributes(self):
151			try:
152			attributes = self._get_property('attr_requested')
153			if not attributes:
154			return []
155			return attributes['required']
156			except Exception:
157			return []
158
159			@property
160			def attributes_optional(self):
161			try:
162			attributes = self._get_property('attr_requested')
163			if not attributes:
164			return []
165			return attributes['optional']
166			except Exception:
167			return []
168
169			@property
170			def organization(self):
171			organization = self._get_property('organization')
172			if not organization:
173			return []
174
175			vals = []
176			for lang, data in organization.items():
177			data['lang'] = lang
178			vals.append(data)
179
180			return vals
181
182			@property
183			def display_name(self):
184			try:
185			return self._get_property('displayName')
186			except Exception:
187			return ''
188
189			@property
190			def federations_count(self):
191			try:
192			return str(self.federations.all().count())
193			except Exception:
194			return ''
195
196			@property
197			def description(self):
198			try:
199			return self._get_property('description')
200			except Exception:
201			return ''
202
203			@property
204			def info_url(self):
205			try:
206			return self._get_property('infoUrl')
207			except Exception:
208			return ''
209
210			@property
211			def privacy_url(self):
212			try:
213			return self._get_property('privacyUrl')
214			except Exception:
215			return ''
216
217			@property
218			def xml(self):
219			try:
220			return self._get_property('xml')
221			except Exception:
222			return ''
223
224			@property
225			def xml_types(self):
226			try:
227			return self._get_property('entity_types')
228			except Exception:
229			return []
230
231			@property
232			def xml_categories(self):
233			try:
234			return self._get_property('entity_categories')
235			except Exception:
236			return []
237
238			@property
239			def display_protocols(self):
240			protocols = []
241
242			#xml_protocols = self._get_property('protocols')
243			xml_protocols = self._display_protocols
244			if xml_protocols:
245			for proto in xml_protocols.split(' '):
246			protocols.append(self.READABLE_PROTOCOLS.get(proto, proto))
247
248			return protocols
249
250			def display_attributes(self):
251			attributes = {}
252			for [attr, friendly] in self.attributes:
253			if friendly:
254			attributes[attr] = friendly
255			elif attr in attributemap.MAP['fro']:
256			attributes[attr] = attributemap.MAP['fro'][attr]
257			else:
258			attributes[attr] = '?'
259			return attributes
260
261			def display_attributes_optional(self):
262			attributes = {}
263			for [attr, friendly] in self.attributes_optional:
264			if friendly:
265			attributes[attr] = friendly
266			elif attr in attributemap.MAP['fro']:
267			attributes[attr] = attributemap.MAP['fro'][attr]
268			else:
269			attributes[attr] = '?'
270			return attributes
271
272			@property
273			def contacts(self):
274			contacts = []
275			for cur_contact in self._get_property('contacts'):
276			if cur_contact['name'] and cur_contact['surname']:
277			contact_name = '%s %s' % (cur_contact['name'], cur_contact['surname'])
278			elif cur_contact['name']:
279			contact_name = cur_contact['name']
280			elif cur_contact['surname']:
281			contact_name = cur_contact['surname']
282			else:
283			contact_name = urlparse(cur_contact['email']).path.partition('?')[0]
284			c_type = 'undefined'
285			if cur_contact['type']:
286			c_type = cur_contact['type']
287			contacts.append({ 'name': contact_name, 'email': cur_contact['email'], 'type': c_type })
288			return contacts
289
290			@property
291			def logos(self):
292			logos = []
293			for cur_logo in self._get_property('logos'):
294			cur_logo['external'] = True
295			logos.append(cur_logo)
296
297			return logos
298
299			class Meta(object):
300			verbose_name = _(u'Entity')
301			verbose_name_plural = _(u'Entities')
302
303			def __unicode__(self):
304			return self.entityid
305
306			def load_metadata(self, federation=None, entity_data=None):
307			if hasattr(self, '_entity_cached'):
308			return
309
310			if self.file:
311			self._entity_cached = self.load_file().get_entity(self.entityid)
312			elif federation:
313			self._entity_cached = federation.get_entity_metadata(self.entityid)
314			elif entity_data:
315			self._entity_cached = entity_data
316			else:
317			right_fed = None
318			first_fed = None
319			for fed in self.federations.all():
320			if fed.registration_authority == self.registration_authority:
321			right_fed = fed
322			if first_fed is None:
323			first_fed = fed
324
325			if right_fed is not None:
326			entity_cached = right_fed.get_entity_metadata(self.entityid)
327			self._entity_cached = entity_cached
328			else:
329			entity_cached = first_fed.get_entity_metadata(self.entityid)
330			self._entity_cached = entity_cached
331
332			if not hasattr(self, '_entity_cached'):
333			raise ValueError("Can't find entity metadata")
334
335			def _get_property(self, prop, federation=None):
336			try:
337			self.load_metadata(federation or self.curfed)
338			except ValueError:
339			return None
340
341			if hasattr(self, '_entity_cached'):
342			return self._entity_cached.get(prop, None)
343			else:
344			raise ValueError("Not metadata loaded")
345
346			def _get_or_create_etypes(self, cached_entity_types):
347			entity_types = []
348			cur_cached_types = [t.xmlname for t in self.types.all()]
349			for etype in self.xml_types:
350			if etype in cur_cached_types:
351			break
352
353			if cached_entity_types is None:
354			entity_type, _ = EntityType.objects.get_or_create(xmlname=etype,
355			name=DESCRIPTOR_TYPES_DISPLAY[etype])
356			else:
357			if etype in cached_entity_types:
358			entity_type = cached_entity_types[etype]
359			else:
360			entity_type = EntityType.objects.create(xmlname=etype,
361			name=DESCRIPTOR_TYPES_DISPLAY[etype])
362			entity_types.append(entity_type)
363			return entity_types
364
365			def _get_or_create_ecategories(self, cached_entity_categories):
366			entity_categories = []
367			cur_cached_categories = [t.category_id for t in self.entity_categories.all()]
368			for ecategory in self.xml_categories:
369			if ecategory in cur_cached_categories:
370			break
371
372			if cached_entity_categories is None:
373			entity_category, _ = EntityCategory.objects.get_or_create(category_id=ecategory)
374			else:
375			if ecategory in cached_entity_categories:
376			entity_category = cached_entity_categories[ecategory]
377			else:
378			entity_category = EntityCategory.objects.create(category_id=ecategory)
379			entity_categories.append(entity_category)
380			return entity_categories
381
382			def process_metadata(self, auto_save=True, entity_data=None, cached_entity_types=None):
383			if not entity_data:
384			self.load_metadata()
385
386			if self.entityid.lower() != entity_data.get('entityid').lower():
387			raise ValueError("EntityID is not the same: %s != %s" % (self.entityid.lower(), entity_data.get('entityid').lower()))
388
389			self._entity_cached = entity_data
390
391			if self.xml_types:
392			entity_types = self._get_or_create_etypes(cached_entity_types)
393			if len(entity_types) > 0:
394			self.types.add(*entity_types)
395
396			if self.xml_categories:
397			db_entity_categories = EntityCategory.objects.all()
398			cached_entity_categories = { entity_category.category_id: entity_category for entity_category in db_entity_categories }
399
400			# Delete categories no more present in XML
401			self.entity_categories.clear()
402
403			# Create all entities, if not alread in database
404			entity_categories = self._get_or_create_ecategories(cached_entity_categories)
405
406			# Add categories to entity
407			if len(entity_categories) > 0:
408			self.entity_categories.add(*entity_categories)
409			else:
410			# No categories in XML, delete eventual categorie sin DB
411			self.entity_categories.clear()
412
413			newname = self._get_property('displayName')
414			if newname and newname != '':
415			self.name = newname
416
417			newprotocols = self.protocols
418			if newprotocols and newprotocols != "":
419			self._display_protocols = newprotocols
420
421			self.certstats = self._get_property('certstats')
422
423			if str(self._get_property('registration_authority')) != '':
424			self.registration_authority = self._get_property('registration_authority')
425
426			if auto_save:
427			self.save()
428
429			def to_dict(self):
430			self.load_metadata()
431
432			entity = self._entity_cached.copy()
433			entity["types"] = [unicode(f) for f in self.types.all()]
434			entity["federations"] = [{u"name": unicode(f), u"url": f.get_absolute_url()}
435			for f in self.federations.all()]
436
437			if self.registration_authority:
438			entity["registration_authority"] = self.registration_authority
439			if self.registration_instant:
440			entity["registration_instant"] = '%s' % self.registration_instant
441
442			if "file_id" in entity.keys():
443			del entity["file_id"]
444			if "entity_types" in entity.keys():
445			del entity["entity_types"]
446
447			return entity
448
449			def display_etype(value, separator=', '):
450			return separator.join([unicode(item) for item in value.all()])
451
452			@classmethod
453			def get_most_federated_entities(self, maxlength=TOP_LENGTH, cache_expire=None):
454			entities = None
455			if cache_expire:
456			entities = cache.get("most_federated_entities")
457
458			if not entities or len(entities) < maxlength:
459			# Entities with count how many federations belongs to, and sorted by most first
460			ob_entities = Entity.objects.all().annotate(federationslength=Count("federations")).order_by("-federationslength")
461			ob_entities = ob_entities.prefetch_related('types', 'federations')
462			ob_entities = ob_entities[:maxlength]
463
464			entities = []
465			for entity in ob_entities:
466			entities.append({
467			'entityid': entity.entityid,
468			'name': entity.name,
469			'absolute_url': entity.get_absolute_url(),
470			'types': [unicode(item) for item in entity.types.all()],
471			'federations': [(unicode(item.name), item.get_absolute_url()) for item in entity.federations.all()],
472			})
473
474			if cache_expire:
475			cache.set("most_federated_entities", entities, cache_expire)
476
477			return entities[:maxlength]
478
479			def get_absolute_url(self):
480			return reverse('entity_view', args=[quote_plus(self.entityid.encode('utf-8'))])
481
482			def can_edit(self, user, delete):
483			permission = 'delete_entity' if delete else 'change_entity'
484			if user.is_superuser or (user.has_perm('metadataparser.%s' % permission) and user in self.editor_users.all()):
485			return True
486
487			for federation in self.federations.all():
488			if federation.can_edit(user, False):
489			return True
490
491			return False
492
493			def has_changed(self, entityid, name, registration_authority, certstats, display_protocols):
494			if self.entityid != entityid:
495			return True
496			if self.name != name:
497			return True
498			if self.registration_authority != registration_authority:
499			return True
500			if self.certstats != certstats:
501			return True
502			if self._display_protocols != display_protocols:
503			return True
504
505			return False
506

GEANT / met

Push — master ( 7e5f4d...79309a )

Entity.has_changed() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like