28592942691

Committed 02 Jul 2026 01:15PM UTC coverage: 86.732% (+0.2%) from 86.582%

Build # 28592942691

Build Type

Pull #1007

github

Committed by

web-flow

Commit Message

Merge b6de47e36 into 5e2ad3f63

Pull Request Pull Request #1007: Data Noice model for information about datasets

Coverage Stats

8838 of 10190 relevant lines covered (86.73%)

0.87 hits per line

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

83.94

alyx/data/admin.py

from django.db.models import Count, Exists, OuterRef, ProtectedError
from django.contrib import admin, messages
from django.utils.html import format_html
from django_admin_listfilter_dropdown.filters import (
    RelatedDropdownFilter,
    ChoiceDropdownFilter,
    SimpleDropdownFilter,
)
from rangefilter.filters import DateRangeFilter

from .models import (DataRepositoryType, DataRepository, DataFormat, DatasetType,
                     Dataset, FileRecord, Download, Revision, Tag, DataNotice)
from alyx.base import BaseAdmin, BaseInlineAdmin, DefaultListFilter, get_admin_url


class CreatedByListFilter(DefaultListFilter):
    title = 'created by'
    parameter_name = 'created_by'

    def lookups(self, request, model_admin):
        return (
            (None, 'Me'),
            ('all', 'All'),
        )

    def queryset(self, request, queryset):
        if self.value() is None:
            return queryset.filter(created_by=request.user)
        elif self.value == 'all':
            return queryset.all()


class DataRepositoryTypeAdmin(BaseAdmin):
    fields = ('name', 'json')
    list_display = ('name',)
    ordering = ('name',)


class DataRepositoryAdmin(BaseAdmin):
    fields = ('name', 'repository_type', 'timezone', 'hostname', 'data_url', 'globus_path',
              'globus_endpoint_id', 'globus_is_personal')
    list_display = fields
    ordering = ('name',)


class DataFormatAdmin(BaseAdmin):
    fields = ['name', 'description', 'file_extension',
              'matlab_loader_function', 'python_loader_function']
    list_display = fields[:-1]
    ordering = ('name',)


class DatasetTypeAdmin(BaseAdmin):
    fields = ('name', 'description', 'filename_pattern', 'created_by')
    list_display = ('name', 'fcount', 'description', 'filename_pattern', 'created_by')
    ordering = ('name',)
    search_fields = ('name', 'description', 'filename_pattern', 'created_by__username')
    list_filter = [('created_by', RelatedDropdownFilter)]

    def get_queryset(self, request):
        qs = super(DatasetTypeAdmin, self).get_queryset(request)
        qs = qs.select_related('created_by')
        return qs

    def save_model(self, request, obj, form, change):
        if not obj.created_by and 'created_by' not in form.changed_data:
            obj.created_by = request.user
        super(DatasetTypeAdmin, self).save_model(request, obj, form, change)

    def fcount(self, dt):
        return Dataset.objects.filter(dataset_type=dt).count()


class BaseExperimentalDataAdmin(BaseAdmin):
    def __init__(self, *args, **kwargs):
        for field in ('created_by', 'created_datetime'):
            if self.fields and field not in self.fields:
                self.fields += (field,)
        super(BaseAdmin, self).__init__(*args, **kwargs)


class FileRecordInline(BaseInlineAdmin):
    model = FileRecord
    extra = 1
    fields = ('data_repository', 'relative_path', 'exists')


class DatasetAdmin(BaseExperimentalDataAdmin):
    fields = ['name', '_online', 'version', 'dataset_type', 'file_size', 'hash',
              'session_ro', 'collection', 'auto_datetime', 'revision_', 'default_dataset',
              '_protected', '_public', 'tags', 'qc']
    readonly_fields = ['name_', 'session_ro', '_online', 'auto_datetime', 'revision_',
                       '_protected', '_public', 'tags', 'qc']
    list_display = ['name_', '_online', 'version', 'collection', 'dataset_type_', 'file_size',
                    'session_ro', 'created_by', 'created_datetime', 'qc']
    inlines = [FileRecordInline]
    list_filter = [('created_by', RelatedDropdownFilter),
                   ('created_datetime', DateRangeFilter),
                   ('dataset_type', RelatedDropdownFilter),
                   ('tags', RelatedDropdownFilter),
                   ('qc', ChoiceDropdownFilter)
                   ]
    search_fields = ('session__id', 'name', 'collection', 'dataset_type__name',
                     'dataset_type__filename_pattern', 'version')
    ordering = ('-created_datetime',)

    def get_queryset(self, request):
        queryset = super(DatasetAdmin, self).get_queryset(request)
        queryset = queryset.select_related('session', 'session__subject', 'created_by')
        return queryset

    def dataset_type_(self, obj):
        return obj.dataset_type.name

    def name_(self, obj):
        return obj.name or '<unnamed>'

    def revision_(self, obj):
        return obj.revision.name

    def session_ro(self, obj):
        url = get_admin_url(obj.session)
        return format_html('<a href="{url}">{name}</a>', url=url, name=obj.session)
    session_ro.short_description = 'session'

    def subject(self, obj):
        return obj.session.subject.nickname

    def _online(self, obj):
        return obj.is_online
    _online.short_description = 'On server'
    _online.boolean = True

    def _protected(self, obj):
        return obj.is_protected
    _protected.short_description = 'Protected'
    _protected.boolean = True

    def _public(self, obj):
        return obj.is_public
    _public.short_description = 'Public'
    _public.boolean = True

    def delete_queryset(self, request, queryset):
        try:
            queryset.delete()
        except ProtectedError as e:
            err_msg = e.args[0] if e.args else 'One or more dataset(s) protected'
            self.message_user(request, err_msg, level=messages.ERROR)

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


class FileRecordAdmin(BaseAdmin):
    fields = ('relative_path', 'data_repository', 'dataset', 'exists')
    list_display = ('relative_path', 'repository', 'dataset_name',
                    'user', 'datetime', 'exists')
    readonly_fields = ('dataset', 'dataset_name', 'repository', 'user', 'datetime')
    list_filter = ('exists', 'data_repository__name')
    search_fields = ('dataset__created_by__username', 'dataset__name',
                     'relative_path', 'data_repository__name')
    ordering = ('-dataset__created_datetime',)

    def get_queryset(self, request):
        qs = super(FileRecordAdmin, self).get_queryset(request)
        qs = qs.select_related('data_repository', 'dataset', 'dataset__created_by')
        return qs

    def repository(self, obj):
        return getattr(obj.data_repository, 'name', None)

    def dataset_name(self, obj):
        return getattr(obj.dataset, 'name', None)

    def user(self, obj):
        return getattr(obj.dataset, 'created_by', None)

    def datetime(self, obj):
        return getattr(obj.dataset, 'created_datetime', None)


class DownloadAdmin(BaseAdmin):
    fields = ('user', 'dataset', 'first_download', 'last_download', 'count', 'projects')
    autocomplete_fields = ('dataset',)
    readonly_fields = ('first_download', 'last_download')
    list_display = ('dataset_type', 'dataset_name', 'subject', 'created_by',
                    'user', 'first_download', 'last_download', 'count')
    list_display_links = ('first_download',)
    search_filter = ('user__username', 'dataset__name')

    def dataset_name(self, obj):
        return obj.dataset.name

    def subject(self, obj):
        return obj.dataset.session.subject.nickname

    def dataset_type(self, obj):
        return obj.dataset.dataset_type.name

    def created_by(self, obj):
        return obj.dataset.created_by.username


class RevisionAdmin(BaseAdmin):
    fields = ['name', 'description', 'created_datetime']
    readonly_fields = ['created_datetime']
    list_display = ['name', 'description']
    search_fields = ('name',)
    ordering = ('-created_datetime',)


class TagAdmin(BaseAdmin):
    fields = ['name', 'description', 'protected', 'public', 'dataset_count', 'session_count']
    list_display = ['name', 'description', 'dataset_count', 'session_count', 'protected', 'public']
    readonly_fields = ['dataset_count', 'session_count']
    search_fields = ('name',)
    ordering = ('name',)

    def dataset_count(self, tag):
        return tag.dataset_count

    def session_count(self, tag):
        return tag.session_count

    def get_queryset(self, request):
        queryset = super().get_queryset(request)
        queryset = queryset.annotate(dataset_count=Count("datasets"))
        queryset = queryset.annotate(session_count=Count("datasets__session", distinct=True))
        return queryset


class DataNoticeAdmin(BaseAdmin):
    fields = (
        'name',
        'description',
        'importance',
        'version_affected',
        'affected_date_start',
        'affected_date_end',
        'datasets',
        'created_by',
        'created_datetime',
        'json',
    )
    readonly_fields = ('created_datetime',)

    class DatasetTagListFilter(SimpleDropdownFilter):
        title = 'dataset tag'
        parameter_name = 'dataset_tag'

        def lookups(self, request, model_admin):
            return Tag.objects.order_by('name').values_list('id', 'name')

        def queryset(self, request, queryset):
            """Filter DataNotice queryset by dataset tag.

            This filter avoids joining the full Dataset table directly.
            """
            value = self.value()
            if not value:
                return queryset

            notice_dataset_through = DataNotice.datasets.through
            dataset_tag_through = Dataset.tags.through

            matching_datasets = dataset_tag_through.objects.filter(tag_id=value).values('dataset_id')
            matching_notice_datasets = notice_dataset_through.objects.filter(
                datanotice_id=OuterRef('pk'), dataset_id__in=matching_datasets)

            return queryset.annotate(_has_dataset_tag=Exists(matching_notice_datasets)).filter(
                _has_dataset_tag=True)

    def has_change_permission(self, request, obj=None):
        # DataNotice has no subject/session ownership; any non-public authenticated user may edit.
        if request.user.is_public_user:
            return False
        return True

    list_display = (
        'name',
        'importance',
        'version_affected',
        'affected_date_start',
        'affected_date_end',
        'created_by',
        'created_datetime',
    )
    list_filter = (DatasetTagListFilter,)
    search_fields = ('name', 'description', 'version_affected', 'created_by__username')
    autocomplete_fields = ('datasets',)


admin.site.register(DataRepositoryType, DataRepositoryTypeAdmin)
admin.site.register(DataRepository, DataRepositoryAdmin)
admin.site.register(DataFormat, DataFormatAdmin)
admin.site.register(DatasetType, DatasetTypeAdmin)
admin.site.register(Dataset, DatasetAdmin)
admin.site.register(FileRecord, FileRecordAdmin)
admin.site.register(Download, DownloadAdmin)
admin.site.register(Revision, RevisionAdmin)
admin.site.register(Tag, TagAdmin)
admin.site.register(DataNotice, DataNoticeAdmin)

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 .models import (DataRepositoryType, DataRepository, DataFormat, DatasetType,	1✔
12	Dataset, FileRecord, Download, Revision, Tag, DataNotice)
13	from alyx.base import BaseAdmin, BaseInlineAdmin, DefaultListFilter, get_admin_url	1✔
14
15
16	class CreatedByListFilter(DefaultListFilter):	1✔
17	title = 'created by'	1✔
18	parameter_name = 'created_by'	1✔
19
20	def lookups(self, request, model_admin):	1✔
21	return (	×
22	(None, 'Me'),
23	('all', 'All'),
24	)
25
26	def queryset(self, request, queryset):	1✔
27	if self.value() is None:	×
28	return queryset.filter(created_by=request.user)	×
29	elif self.value == 'all':	×
30	return queryset.all()	×
31
32
33	class DataRepositoryTypeAdmin(BaseAdmin):	1✔
34	fields = ('name', 'json')	1✔
35	list_display = ('name',)	1✔
36	ordering = ('name',)	1✔
37
38
39	class DataRepositoryAdmin(BaseAdmin):	1✔
40	fields = ('name', 'repository_type', 'timezone', 'hostname', 'data_url', 'globus_path',	1✔
41	'globus_endpoint_id', 'globus_is_personal')
42	list_display = fields	1✔
43	ordering = ('name',)	1✔
44
45
46	class DataFormatAdmin(BaseAdmin):	1✔
47	fields = ['name', 'description', 'file_extension',	1✔
48	'matlab_loader_function', 'python_loader_function']
49	list_display = fields[:-1]	1✔
50	ordering = ('name',)	1✔
51
52
53	class DatasetTypeAdmin(BaseAdmin):	1✔
54	fields = ('name', 'description', 'filename_pattern', 'created_by')	1✔
55	list_display = ('name', 'fcount', 'description', 'filename_pattern', 'created_by')	1✔
56	ordering = ('name',)	1✔
57	search_fields = ('name', 'description', 'filename_pattern', 'created_by__username')	1✔
58	list_filter = [('created_by', RelatedDropdownFilter)]	1✔
59
60	def get_queryset(self, request):	1✔
61	qs = super(DatasetTypeAdmin, self).get_queryset(request)	1✔
62	qs = qs.select_related('created_by')	1✔
63	return qs	1✔
64
65	def save_model(self, request, obj, form, change):	1✔
66	if not obj.created_by and 'created_by' not in form.changed_data:	×
67	obj.created_by = request.user	×
68	super(DatasetTypeAdmin, self).save_model(request, obj, form, change)	×
69
70	def fcount(self, dt):	1✔
71	return Dataset.objects.filter(dataset_type=dt).count()	1✔
72
73
74	class BaseExperimentalDataAdmin(BaseAdmin):	1✔
75	def __init__(self, args, *kwargs):	1✔
76	for field in ('created_by', 'created_datetime'):	1✔
77	if self.fields and field not in self.fields:	1✔
78	self.fields += (field,)	1✔
79	super(BaseAdmin, self).__init__(args, *kwargs)	1✔
80
81
82	class FileRecordInline(BaseInlineAdmin):	1✔
83	model = FileRecord	1✔
84	extra = 1	1✔
85	fields = ('data_repository', 'relative_path', 'exists')	1✔
86
87
88	class DatasetAdmin(BaseExperimentalDataAdmin):	1✔
89	fields = ['name', '_online', 'version', 'dataset_type', 'file_size', 'hash',	1✔
90	'session_ro', 'collection', 'auto_datetime', 'revision_', 'default_dataset',
91	'_protected', '_public', 'tags', 'qc']
92	readonly_fields = ['name_', 'session_ro', '_online', 'auto_datetime', 'revision_',	1✔
93	'_protected', '_public', 'tags', 'qc']
94	list_display = ['name_', '_online', 'version', 'collection', 'dataset_type_', 'file_size',	1✔
95	'session_ro', 'created_by', 'created_datetime', 'qc']
96	inlines = [FileRecordInline]	1✔
97	list_filter = [('created_by', RelatedDropdownFilter),	1✔
98	('created_datetime', DateRangeFilter),
99	('dataset_type', RelatedDropdownFilter),
100	('tags', RelatedDropdownFilter),
101	('qc', ChoiceDropdownFilter)
102	]
103	search_fields = ('session__id', 'name', 'collection', 'dataset_type__name',	1✔
104	'dataset_type__filename_pattern', 'version')
105	ordering = ('-created_datetime',)	1✔
106
107	def get_queryset(self, request):	1✔
108	queryset = super(DatasetAdmin, self).get_queryset(request)	1✔
109	queryset = queryset.select_related('session', 'session__subject', 'created_by')	1✔
110	return queryset	1✔
111
112	def dataset_type_(self, obj):	1✔
113	return obj.dataset_type.name	×
114
115	def name_(self, obj):	1✔
116	return obj.name or '<unnamed>'	×
117
118	def revision_(self, obj):	1✔
119	return obj.revision.name	1✔
120
121	def session_ro(self, obj):	1✔
122	url = get_admin_url(obj.session)	1✔
123	return format_html('<a href="{url}">{name}</a>', url=url, name=obj.session)	1✔
124	session_ro.short_description = 'session'	1✔
125
126	def subject(self, obj):	1✔
127	return obj.session.subject.nickname	×
128
129	def _online(self, obj):	1✔
130	return obj.is_online	1✔
131	_online.short_description = 'On server'	1✔
132	_online.boolean = True	1✔
133
134	def _protected(self, obj):	1✔
135	return obj.is_protected	1✔
136	_protected.short_description = 'Protected'	1✔
137	_protected.boolean = True	1✔
138
139	def _public(self, obj):	1✔
140	return obj.is_public	1✔
141	_public.short_description = 'Public'	1✔
142	_public.boolean = True	1✔
143
144	def delete_queryset(self, request, queryset):	1✔
145	try:	×
146	queryset.delete()	×
147	except ProtectedError as e:	×
148	err_msg = e.args[0] if e.args else 'One or more dataset(s) protected'	×
149	self.message_user(request, err_msg, level=messages.ERROR)	×
150
151	def delete_model(self, request, obj):	1✔
152	try:	×
153	obj.delete()	×
154	except ProtectedError as e:	×
155	# FIXME This still shows the successful message which is confusing for users
156	err_msg = e.args[0] if e.args else f'Dataset {obj.name} is protected'	×
157	self.message_user(request, err_msg, level=messages.ERROR)	×
158
159
160	class FileRecordAdmin(BaseAdmin):	1✔
161	fields = ('relative_path', 'data_repository', 'dataset', 'exists')	1✔
162	list_display = ('relative_path', 'repository', 'dataset_name',	1✔
163	'user', 'datetime', 'exists')
164	readonly_fields = ('dataset', 'dataset_name', 'repository', 'user', 'datetime')	1✔
165	list_filter = ('exists', 'data_repository__name')	1✔
166	search_fields = ('dataset__created_by__username', 'dataset__name',	1✔
167	'relative_path', 'data_repository__name')
168	ordering = ('-dataset__created_datetime',)	1✔
169
170	def get_queryset(self, request):	1✔
171	qs = super(FileRecordAdmin, self).get_queryset(request)	1✔
172	qs = qs.select_related('data_repository', 'dataset', 'dataset__created_by')	1✔
173	return qs	1✔
174
175	def repository(self, obj):	1✔
176	return getattr(obj.data_repository, 'name', None)	×
177
178	def dataset_name(self, obj):	1✔
179	return getattr(obj.dataset, 'name', None)	×
180
181	def user(self, obj):	1✔
182	return getattr(obj.dataset, 'created_by', None)	×
183
184	def datetime(self, obj):	1✔
185	return getattr(obj.dataset, 'created_datetime', None)	×
186
187
188	class DownloadAdmin(BaseAdmin):	1✔
189	fields = ('user', 'dataset', 'first_download', 'last_download', 'count', 'projects')	1✔
190	autocomplete_fields = ('dataset',)	1✔
191	readonly_fields = ('first_download', 'last_download')	1✔
192	list_display = ('dataset_type', 'dataset_name', 'subject', 'created_by',	1✔
193	'user', 'first_download', 'last_download', 'count')
194	list_display_links = ('first_download',)	1✔
195	search_filter = ('user__username', 'dataset__name')	1✔
196
197	def dataset_name(self, obj):	1✔
198	return obj.dataset.name	×
199
200	def subject(self, obj):	1✔
201	return obj.dataset.session.subject.nickname	×
202
203	def dataset_type(self, obj):	1✔
204	return obj.dataset.dataset_type.name	×
205
206	def created_by(self, obj):	1✔
207	return obj.dataset.created_by.username	×
208
209
210	class RevisionAdmin(BaseAdmin):	1✔
211	fields = ['name', 'description', 'created_datetime']	1✔
212	readonly_fields = ['created_datetime']	1✔
213	list_display = ['name', 'description']	1✔
214	search_fields = ('name',)	1✔
215	ordering = ('-created_datetime',)	1✔
216
217
218	class TagAdmin(BaseAdmin):	1✔
219	fields = ['name', 'description', 'protected', 'public', 'dataset_count', 'session_count']	1✔
220	list_display = ['name', 'description', 'dataset_count', 'session_count', 'protected', 'public']	1✔
221	readonly_fields = ['dataset_count', 'session_count']	1✔
222	search_fields = ('name',)	1✔
223	ordering = ('name',)	1✔
224
225	def dataset_count(self, tag):	1✔
226	return tag.dataset_count	1✔
227
228	def session_count(self, tag):	1✔
229	return tag.session_count	1✔
230
231	def get_queryset(self, request):	1✔
232	queryset = super().get_queryset(request)	1✔
233	queryset = queryset.annotate(dataset_count=Count("datasets"))	1✔
234	queryset = queryset.annotate(session_count=Count("datasets__session", distinct=True))	1✔
235	return queryset	1✔
236
237
238	class DataNoticeAdmin(BaseAdmin):	1✔
239	fields = (	1✔
240	'name',
241	'description',
242	'importance',
243	'version_affected',
244	'affected_date_start',
245	'affected_date_end',
246	'datasets',
247	'created_by',
248	'created_datetime',
249	'json',
250	)
251	readonly_fields = ('created_datetime',)	1✔
252
253	class DatasetTagListFilter(SimpleDropdownFilter):	1✔
254	title = 'dataset tag'	1✔
255	parameter_name = 'dataset_tag'	1✔
256
257	def lookups(self, request, model_admin):	1✔
258	return Tag.objects.order_by('name').values_list('id', 'name')	1✔
259
260	def queryset(self, request, queryset):	1✔
261	"""Filter DataNotice queryset by dataset tag.
262
263	This filter avoids joining the full Dataset table directly.
264	"""
265	value = self.value()	1✔
266	if not value:	1✔
267	return queryset	×
268
269	notice_dataset_through = DataNotice.datasets.through	1✔
270	dataset_tag_through = Dataset.tags.through	1✔
271
272	matching_datasets = dataset_tag_through.objects.filter(tag_id=value).values('dataset_id')	1✔
273	matching_notice_datasets = notice_dataset_through.objects.filter(	1✔
274	datanotice_id=OuterRef('pk'), dataset_id__in=matching_datasets)
275
276	return queryset.annotate(_has_dataset_tag=Exists(matching_notice_datasets)).filter(	1✔
277	_has_dataset_tag=True)
278
279	def has_change_permission(self, request, obj=None):	1✔
280	# DataNotice has no subject/session ownership; any non-public authenticated user may edit.
281	if request.user.is_public_user:	1✔
282	return False	×
283	return True	1✔
284
285	list_display = (	1✔
286	'name',
287	'importance',
288	'version_affected',
289	'affected_date_start',
290	'affected_date_end',
291	'created_by',
292	'created_datetime',
293	)
294	list_filter = (DatasetTagListFilter,)	1✔
295	search_fields = ('name', 'description', 'version_affected', 'created_by__username')	1✔
296	autocomplete_fields = ('datasets',)	1✔
297
298
299	admin.site.register(DataRepositoryType, DataRepositoryTypeAdmin)	1✔
300	admin.site.register(DataRepository, DataRepositoryAdmin)	1✔
301	admin.site.register(DataFormat, DataFormatAdmin)	1✔
302	admin.site.register(DatasetType, DatasetTypeAdmin)	1✔
303	admin.site.register(Dataset, DatasetAdmin)	1✔
304	admin.site.register(FileRecord, FileRecordAdmin)	1✔
305	admin.site.register(Download, DownloadAdmin)	1✔
306	admin.site.register(Revision, RevisionAdmin)	1✔
307	admin.site.register(Tag, TagAdmin)	1✔
308	admin.site.register(DataNotice, DataNoticeAdmin)	1✔

cortex-lab / alyx / 28592942691

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

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