• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

cortex-lab / alyx / 29997028058

23 Jul 2026 09:51AM UTC coverage: 86.879% (+0.3%) from 86.595%
29997028058

push

github

web-flow
Merge pull request #1022 from cortex-lab/dev

Release 3.6.0

8965 of 10319 relevant lines covered (86.88%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.58
alyx/data/admin.py
1
from django.db.models import Count, Exists, OuterRef, ProtectedError
1✔
2
from django.contrib import admin, messages
1✔
3
from django.utils.html import format_html
1✔
4
from django_admin_listfilter_dropdown.filters import (
1✔
5
    RelatedDropdownFilter,
6
    ChoiceDropdownFilter,
7
    SimpleDropdownFilter,
8
)
9
from rangefilter.filters import DateRangeFilter
1✔
10

11
from actions.models import Session
1✔
12
from subjects.models import Project
1✔
13
from .models import (DataRepositoryType, DataRepository, DataFormat, DatasetType,
1✔
14
                     Dataset, FileRecord, Download, Revision, Tag, DataNotice)
15
from alyx.base import BaseAdmin, BaseInlineAdmin, DefaultListFilter, get_admin_url
1✔
16

17

18
class CreatedByListFilter(DefaultListFilter):
1✔
19
    title = 'created by'
1✔
20
    parameter_name = 'created_by'
1✔
21

22
    def lookups(self, request, model_admin):
1✔
23
        return (
×
24
            (None, 'Me'),
25
            ('all', 'All'),
26
        )
27

28
    def queryset(self, request, queryset):
1✔
29
        if self.value() is None:
×
30
            return queryset.filter(created_by=request.user)
×
31
        elif self.value == 'all':
×
32
            return queryset.all()
×
33

34

35
class DataRepositoryTypeAdmin(BaseAdmin):
1✔
36
    fields = ('name', 'json')
1✔
37
    list_display = ('name',)
1✔
38
    ordering = ('name',)
1✔
39

40

41
class DataRepositoryAdmin(BaseAdmin):
1✔
42
    fields = ('name', 'repository_type', 'timezone', 'hostname', 'data_url', 'globus_path',
1✔
43
              'globus_endpoint_id', 'globus_is_personal')
44
    list_display = fields
1✔
45
    ordering = ('name',)
1✔
46

47

48
class DataFormatAdmin(BaseAdmin):
1✔
49
    fields = ['name', 'description', 'file_extension',
1✔
50
              'matlab_loader_function', 'python_loader_function']
51
    list_display = fields[:-1]
1✔
52
    ordering = ('name',)
1✔
53

54

55
class DatasetTypeAdmin(BaseAdmin):
1✔
56
    fields = ('name', 'description', 'filename_pattern', 'created_by')
1✔
57
    list_display = ('name', 'fcount', 'description', 'filename_pattern', 'created_by')
1✔
58
    ordering = ('name',)
1✔
59
    search_fields = ('name', 'description', 'filename_pattern', 'created_by__username')
1✔
60
    list_filter = [('created_by', RelatedDropdownFilter)]
1✔
61

62
    def get_queryset(self, request):
1✔
63
        qs = super(DatasetTypeAdmin, self).get_queryset(request)
1✔
64
        qs = qs.select_related('created_by')
1✔
65
        return qs
1✔
66

67
    def save_model(self, request, obj, form, change):
1✔
68
        if not obj.created_by and 'created_by' not in form.changed_data:
×
69
            obj.created_by = request.user
×
70
        super(DatasetTypeAdmin, self).save_model(request, obj, form, change)
×
71

72
    def fcount(self, dt):
1✔
73
        return Dataset.objects.filter(dataset_type=dt).count()
×
74

75

76
class BaseExperimentalDataAdmin(BaseAdmin):
1✔
77
    def __init__(self, *args, **kwargs):
1✔
78
        for field in ('created_by', 'created_datetime'):
1✔
79
            if self.fields and field not in self.fields:
1✔
80
                self.fields += (field,)
1✔
81
        super(BaseAdmin, self).__init__(*args, **kwargs)
1✔
82

83

84
class FileRecordInline(BaseInlineAdmin):
1✔
85
    model = FileRecord
1✔
86
    extra = 1
1✔
87
    fields = ('data_repository', 'relative_path', 'exists')
1✔
88

89

90
class DatasetAdmin(BaseExperimentalDataAdmin):
1✔
91
    fields = ['name', '_online', 'version', 'dataset_type', 'file_size', 'hash',
1✔
92
              'session_ro', 'collection', 'auto_datetime', 'revision_', 'default_dataset',
93
              '_protected', '_public', 'tags', 'qc']
94
    readonly_fields = ['name_', 'session_ro', '_online', 'auto_datetime', 'revision_',
1✔
95
                       '_protected', '_public', 'tags', 'qc']
96
    list_display = ['name_', '_online', 'version', 'collection', 'dataset_type_', 'file_size',
1✔
97
                    'session_ro', 'created_by', 'created_datetime', 'qc']
98
    inlines = [FileRecordInline]
1✔
99
    list_filter = [('created_by', RelatedDropdownFilter),
1✔
100
                   ('created_datetime', DateRangeFilter),
101
                   ('dataset_type', RelatedDropdownFilter),
102
                   ('tags', RelatedDropdownFilter),
103
                   ('qc', ChoiceDropdownFilter)
104
                   ]
105
    search_fields = ('session__id', 'name', 'collection', 'dataset_type__name',
1✔
106
                     'dataset_type__filename_pattern', 'version')
107
    ordering = ('-created_datetime',)
1✔
108

109
    def get_queryset(self, request):
1✔
110
        queryset = super(DatasetAdmin, self).get_queryset(request)
1✔
111
        queryset = queryset.select_related('session', 'session__subject', 'created_by')
1✔
112
        return queryset
1✔
113

114
    def dataset_type_(self, obj):
1✔
115
        return obj.dataset_type.name
×
116

117
    def name_(self, obj):
1✔
118
        return obj.name or '<unnamed>'
×
119

120
    def revision_(self, obj):
1✔
121
        return obj.revision.name
1✔
122

123
    def session_ro(self, obj):
1✔
124
        url = get_admin_url(obj.session)
1✔
125
        return format_html('<a href="{url}">{name}</a>', url=url, name=obj.session)
1✔
126
    session_ro.short_description = 'session'
1✔
127

128
    def subject(self, obj):
1✔
129
        return obj.session.subject.nickname
×
130

131
    def _online(self, obj):
1✔
132
        return obj.is_online
1✔
133
    _online.short_description = 'On server'
1✔
134
    _online.boolean = True
1✔
135

136
    def _protected(self, obj):
1✔
137
        return obj.is_protected
1✔
138
    _protected.short_description = 'Protected'
1✔
139
    _protected.boolean = True
1✔
140

141
    def _public(self, obj):
1✔
142
        return obj.is_public
1✔
143
    _public.short_description = 'Public'
1✔
144
    _public.boolean = True
1✔
145

146
    def delete_queryset(self, request, queryset):
1✔
147
        try:
×
148
            queryset.delete()
×
149
        except ProtectedError as e:
×
150
            err_msg = e.args[0] if e.args else 'One or more dataset(s) protected'
×
151
            self.message_user(request, err_msg, level=messages.ERROR)
×
152

153
    def delete_model(self, request, obj):
1✔
154
        try:
×
155
            obj.delete()
×
156
        except ProtectedError as e:
×
157
            # FIXME This still shows the successful message which is confusing for users
158
            err_msg = e.args[0] if e.args else f'Dataset {obj.name} is protected'
×
159
            self.message_user(request, err_msg, level=messages.ERROR)
×
160

161

162
class FileRecordAdmin(BaseAdmin):
1✔
163
    fields = ('relative_path', 'data_repository', 'dataset', 'exists')
1✔
164
    list_display = ('relative_path', 'repository', 'dataset_name',
1✔
165
                    'user', 'datetime', 'exists')
166
    readonly_fields = ('dataset', 'dataset_name', 'repository', 'user', 'datetime')
1✔
167
    list_filter = ('exists', 'data_repository__name')
1✔
168
    search_fields = ('dataset__created_by__username', 'dataset__name',
1✔
169
                     'relative_path', 'data_repository__name')
170
    ordering = ('-dataset__created_datetime',)
1✔
171

172
    def get_queryset(self, request):
1✔
173
        qs = super(FileRecordAdmin, self).get_queryset(request)
1✔
174
        qs = qs.select_related('data_repository', 'dataset', 'dataset__created_by')
1✔
175
        return qs
1✔
176

177
    def repository(self, obj):
1✔
178
        return getattr(obj.data_repository, 'name', None)
×
179

180
    def dataset_name(self, obj):
1✔
181
        return getattr(obj.dataset, 'name', None)
×
182

183
    def user(self, obj):
1✔
184
        return getattr(obj.dataset, 'created_by', None)
×
185

186
    def datetime(self, obj):
1✔
187
        return getattr(obj.dataset, 'created_datetime', None)
×
188

189

190
class DownloadAdmin(BaseAdmin):
1✔
191
    fields = ('user', 'dataset', 'first_download', 'last_download', 'count', 'projects')
1✔
192
    autocomplete_fields = ('dataset',)
1✔
193
    readonly_fields = ('first_download', 'last_download')
1✔
194
    list_display = ('dataset_type', 'dataset_name', 'subject', 'created_by',
1✔
195
                    'user', 'first_download', 'last_download', 'count')
196
    list_display_links = ('first_download',)
1✔
197
    search_filter = ('user__username', 'dataset__name')
1✔
198

199
    def dataset_name(self, obj):
1✔
200
        return obj.dataset.name
×
201

202
    def subject(self, obj):
1✔
203
        return obj.dataset.session.subject.nickname
×
204

205
    def dataset_type(self, obj):
1✔
206
        return obj.dataset.dataset_type.name
×
207

208
    def created_by(self, obj):
1✔
209
        return obj.dataset.created_by.username
×
210

211

212
class RevisionAdmin(BaseAdmin):
1✔
213
    fields = ['name', 'description', 'created_datetime']
1✔
214
    readonly_fields = ['created_datetime']
1✔
215
    list_display = ['name', 'description']
1✔
216
    search_fields = ('name',)
1✔
217
    ordering = ('-created_datetime',)
1✔
218

219

220
class TagAdmin(BaseAdmin):
1✔
221
    fields = ['name', 'description', 'protected', 'public', 'dataset_count', 'session_count']
1✔
222
    list_display = ['name', 'description', 'dataset_count', 'session_count', 'protected', 'public']
1✔
223
    readonly_fields = ['dataset_count', 'session_count']
1✔
224
    search_fields = ('name',)
1✔
225
    ordering = ('name',)
1✔
226

227
    def dataset_count(self, tag):
1✔
228
        return tag.dataset_count
1✔
229

230
    def session_count(self, tag):
1✔
231
        return tag.session_count
1✔
232

233
    def get_queryset(self, request):
1✔
234
        queryset = super().get_queryset(request)
1✔
235
        queryset = queryset.annotate(dataset_count=Count("datasets"))
1✔
236
        queryset = queryset.annotate(session_count=Count("datasets__session", distinct=True))
1✔
237
        return queryset
1✔
238

239

240
class DataNoticeAdmin(BaseAdmin):
1✔
241
    fields = (
1✔
242
        'name',
243
        'description',
244
        'importance',
245
        'version_affected',
246
        'affected_date_start',
247
        'affected_date_end',
248
        'datasets',
249
        'created_by',
250
        'created_datetime',
251
        'json',
252
    )
253
    readonly_fields = ('created_datetime',)
1✔
254

255
    class DatasetTagListFilter(SimpleDropdownFilter):
1✔
256
        title = 'dataset tag'
1✔
257
        parameter_name = 'dataset_tag'
1✔
258

259
        def lookups(self, request, model_admin):
1✔
260
            return Tag.objects.order_by('name').values_list('id', 'name')
1✔
261

262
        def queryset(self, request, queryset):
1✔
263
            """Filter DataNotice queryset by dataset tag.
264

265
            This filter avoids joining the full Dataset table directly.
266
            """
267
            value = self.value()
1✔
268
            if not value:
1✔
269
                return queryset
×
270

271
            notice_dataset_through = DataNotice.datasets.through
1✔
272
            dataset_tag_through = Dataset.tags.through
1✔
273

274
            matching_datasets = dataset_tag_through.objects.filter(tag_id=value).values('dataset_id')
1✔
275
            matching_notice_datasets = notice_dataset_through.objects.filter(
1✔
276
                datanotice_id=OuterRef('pk'), dataset_id__in=matching_datasets)
277

278
            return queryset.annotate(_has_dataset_tag=Exists(matching_notice_datasets)).filter(
1✔
279
                _has_dataset_tag=True)
280

281
    class SessionProjectListFilter(SimpleDropdownFilter):
1✔
282
        title = 'project'
1✔
283
        parameter_name = 'project'
1✔
284

285
        def lookups(self, request, model_admin):
1✔
286
            notice_dataset_through = DataNotice.datasets.through
1✔
287
            session_project_through = Session.projects.through
1✔
288

289
            session_ids = Dataset.objects.filter(
1✔
290
                id__in=notice_dataset_through.objects.values('dataset_id'),
291
                session_id__isnull=False,
292
            ).values('session_id')
293
            project_ids = session_project_through.objects.filter(
1✔
294
                session_id__in=session_ids,
295
            ).values('project_id')
296

297
            return Project.objects.filter(id__in=project_ids).order_by('name').values_list('id', 'name')
1✔
298

299
        def queryset(self, request, queryset):
1✔
300
            value = self.value()
1✔
301
            if not value:
1✔
302
                return queryset
×
303

304
            notice_dataset_through = DataNotice.datasets.through
1✔
305
            session_project_through = Session.projects.through
1✔
306

307
            matching_sessions = session_project_through.objects.filter(project_id=value).values('session_id')
1✔
308
            matching_datasets = Dataset.objects.filter(session_id__in=matching_sessions).values('id')
1✔
309
            matching_notice_datasets = notice_dataset_through.objects.filter(
1✔
310
                datanotice_id=OuterRef('pk'), dataset_id__in=matching_datasets)
311

312
            return queryset.annotate(_has_project=Exists(matching_notice_datasets)).filter(
1✔
313
                _has_project=True)
314

315
    def has_change_permission(self, request, obj=None):
1✔
316
        # DataNotice has no subject/session ownership; any non-public authenticated user may edit.
317
        if request.user.is_public_user:
1✔
318
            return False
×
319
        return True
1✔
320

321
    list_display = (
1✔
322
        'name',
323
        'importance',
324
        'version_affected',
325
        'affected_date_start',
326
        'affected_date_end',
327
        'created_by',
328
        'created_datetime',
329
    )
330
    list_filter = (DatasetTagListFilter, SessionProjectListFilter)
1✔
331
    search_fields = ('name', 'description', 'version_affected', 'created_by__username')
1✔
332
    autocomplete_fields = ('datasets',)
1✔
333

334

335
admin.site.register(DataRepositoryType, DataRepositoryTypeAdmin)
1✔
336
admin.site.register(DataRepository, DataRepositoryAdmin)
1✔
337
admin.site.register(DataFormat, DataFormatAdmin)
1✔
338
admin.site.register(DatasetType, DatasetTypeAdmin)
1✔
339
admin.site.register(Dataset, DatasetAdmin)
1✔
340
admin.site.register(FileRecord, FileRecordAdmin)
1✔
341
admin.site.register(Download, DownloadAdmin)
1✔
342
admin.site.register(Revision, RevisionAdmin)
1✔
343
admin.site.register(Tag, TagAdmin)
1✔
344
admin.site.register(DataNotice, DataNoticeAdmin)
1✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE TRIAL · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc