-
Notifications
You must be signed in to change notification settings - Fork 27
/
models.py
129 lines (107 loc) · 4.4 KB
/
models.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType
from django.db import models
from pydash import get
from core.common.fields import URIField
from core.common.models import BaseModel
class URLRegistry(BaseModel):
url = URIField()
name = models.TextField(null=True, blank=True)
namespace = models.CharField(max_length=300, null=True, blank=True)
organization = models.ForeignKey(
'orgs.Organization', on_delete=models.CASCADE, null=True, blank=True,
related_name='url_registry_entries'
)
user = models.ForeignKey(
'users.UserProfile', on_delete=models.CASCADE, null=True, blank=True,
related_name='url_registry_entries'
)
repo_type = models.ForeignKey(ContentType, on_delete=models.CASCADE, null=True, blank=True)
repo_id = models.PositiveIntegerField(null=True, blank=True)
repo = GenericForeignKey('repo_type', 'repo_id')
public_access = None
uri = None
OBJECT_TYPE = 'URLRegistryEntry'
es_fields = {
'name': {'sortable': False, 'filterable': True, 'exact': True},
'_name': {'sortable': True, 'filterable': False, 'exact': False},
'namespace': {'sortable': False, 'filterable': True, 'exact': True},
'_namespace': {'sortable': True, 'filterable': False, 'exact': False},
'url': {'sortable': False, 'filterable': True, 'exact': True},
'_url': {'sortable': True, 'filterable': False, 'exact': False},
'last_update': {'sortable': True, 'default': 'desc', 'filterable': False},
'updated_by': {'sortable': False, 'filterable': False, 'facet': True}
}
class Meta:
db_table = 'url_registries'
constraints = [
models.UniqueConstraint(
condition=models.Q(is_active=True, user__isnull=False),
fields=('user', 'url'), name='user_url_unique'
),
models.UniqueConstraint(
condition=models.Q(is_active=True, organization__isnull=False),
fields=('organization', 'url'), name='org_url_unique'
),
models.UniqueConstraint(
condition=models.Q(is_active=True, organization__isnull=True, user__isnull=True),
fields=('url',), name='global_url_unique'
),
]
@property
def relative_uri(self):
owner = self.owner
return f'{owner.uri}url-registry/{self.id}/' if owner else f'/url-registry/{self.id}/'
def save(self, *args, **kwargs):
self.clean()
super().save(*args, **kwargs)
@property
def owner(self):
return self.organization or self.user
@property
def owner_url(self):
owner = self.owner
return get(owner, 'uri') or '/'
@property
def owner_type(self):
return get(self.owner, 'resource_type') or None
@property
def active_entries(self):
return URLRegistry.get_active_entries(self.owner)
def is_uniq(self):
return not self.active_entries.filter(url=self.url).exists()
@classmethod
def get_active_entries(cls, owner):
queryset = owner.url_registry_entries if owner else cls.get_global_entries()
return queryset.filter(is_active=True)
@classmethod
def get_global_entries(cls):
return cls.objects.filter(organization__isnull=True, user__isnull=True)
@classmethod
def get_active_global_entries(cls):
return cls.get_global_entries().filter(is_active=True)
@property
def namespace_owner(self):
from core.common.mixins import SourceContainerMixin
return SourceContainerMixin.get_object_from_namespace(self.namespace)
@classmethod
def lookup(cls, url, registry_owner=None):
entry = cls.get_entry(url, registry_owner)
if entry:
return entry.lookup_repo()
return None
def lookup_repo(self):
namespace_owner = self.namespace_owner
if namespace_owner:
repo = namespace_owner.find_repo_by_canonical_url(self.url)
self._set_repo(repo)
return repo
return None
@classmethod
def get_entry(cls, url, owner=None):
entries = owner.url_registry_entries.filter(is_active=True) if owner else cls.get_active_global_entries()
return entries.filter(url=url).first()
def _set_repo(self, repo):
if repo:
self.repo = repo
self.save()