29940300361

Committed 22 Jul 2026 04:57PM UTC coverage: 86.585% (-0.01%) from 86.595%

Build # 29940300361

Build Type

Pull #1021

github

Committed by

web-flow

Commit Message

Merge 9f55d8ba1 into a461acfd5

Pull Request Pull Request #1021: Remove unknown data set type from test dump

Coverage Stats

8636 of 9974 relevant lines covered (86.59%)

0.87 hits per line

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

82.04

alyx/data/admin.py

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

from .models import (DataRepositoryType, DataRepository, DataFormat, DatasetType,
                     Dataset, FileRecord, Download, Revision, Tag)
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


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)

1	from django.db.models import Count, 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 RelatedDropdownFilter, ChoiceDropdownFilter	1✔
5	from rangefilter.filters import DateRangeFilter	1✔
6
7	from .models import (DataRepositoryType, DataRepository, DataFormat, DatasetType,	1✔
8	Dataset, FileRecord, Download, Revision, Tag)
9	from alyx.base import BaseAdmin, BaseInlineAdmin, DefaultListFilter, get_admin_url	1✔
10
11
12	class CreatedByListFilter(DefaultListFilter):	1✔
13	title = 'created by'	1✔
14	parameter_name = 'created_by'	1✔
15
16	def lookups(self, request, model_admin):	1✔
17	return (	×
18	(None, 'Me'),
19	('all', 'All'),
20	)
21
22	def queryset(self, request, queryset):	1✔
23	if self.value() is None:	×
24	return queryset.filter(created_by=request.user)	×
25	elif self.value == 'all':	×
26	return queryset.all()	×
27
28
29	class DataRepositoryTypeAdmin(BaseAdmin):	1✔
30	fields = ('name', 'json')	1✔
31	list_display = ('name',)	1✔
32	ordering = ('name',)	1✔
33
34
35	class DataRepositoryAdmin(BaseAdmin):	1✔
36	fields = ('name', 'repository_type', 'timezone', 'hostname', 'data_url', 'globus_path',	1✔
37	'globus_endpoint_id', 'globus_is_personal')
38	list_display = fields	1✔
39	ordering = ('name',)	1✔
40
41
42	class DataFormatAdmin(BaseAdmin):	1✔
43	fields = ['name', 'description', 'file_extension',	1✔
44	'matlab_loader_function', 'python_loader_function']
45	list_display = fields[:-1]	1✔
46	ordering = ('name',)	1✔
47
48
49	class DatasetTypeAdmin(BaseAdmin):	1✔
50	fields = ('name', 'description', 'filename_pattern', 'created_by')	1✔
51	list_display = ('name', 'fcount', 'description', 'filename_pattern', 'created_by')	1✔
52	ordering = ('name',)	1✔
53	search_fields = ('name', 'description', 'filename_pattern', 'created_by__username')	1✔
54	list_filter = [('created_by', RelatedDropdownFilter)]	1✔
55
56	def get_queryset(self, request):	1✔
57	qs = super(DatasetTypeAdmin, self).get_queryset(request)	1✔
58	qs = qs.select_related('created_by')	1✔
59	return qs	1✔
60
61	def save_model(self, request, obj, form, change):	1✔
62	if not obj.created_by and 'created_by' not in form.changed_data:	×
63	obj.created_by = request.user	×
64	super(DatasetTypeAdmin, self).save_model(request, obj, form, change)	×
65
66	def fcount(self, dt):	1✔
67	return Dataset.objects.filter(dataset_type=dt).count()	×
68
69
70	class BaseExperimentalDataAdmin(BaseAdmin):	1✔
71	def __init__(self, args, *kwargs):	1✔
72	for field in ('created_by', 'created_datetime'):	1✔
73	if self.fields and field not in self.fields:	1✔
74	self.fields += (field,)	1✔
75	super(BaseAdmin, self).__init__(args, *kwargs)	1✔
76
77
78	class FileRecordInline(BaseInlineAdmin):	1✔
79	model = FileRecord	1✔
80	extra = 1	1✔
81	fields = ('data_repository', 'relative_path', 'exists')	1✔
82
83
84	class DatasetAdmin(BaseExperimentalDataAdmin):	1✔
85	fields = ['name', '_online', 'version', 'dataset_type', 'file_size', 'hash',	1✔
86	'session_ro', 'collection', 'auto_datetime', 'revision_', 'default_dataset',
87	'_protected', '_public', 'tags', 'qc']
88	readonly_fields = ['name_', 'session_ro', '_online', 'auto_datetime', 'revision_',	1✔
89	'_protected', '_public', 'tags', 'qc']
90	list_display = ['name_', '_online', 'version', 'collection', 'dataset_type_', 'file_size',	1✔
91	'session_ro', 'created_by', 'created_datetime', 'qc']
92	inlines = [FileRecordInline]	1✔
93	list_filter = [('created_by', RelatedDropdownFilter),	1✔
94	('created_datetime', DateRangeFilter),
95	('dataset_type', RelatedDropdownFilter),
96	('tags', RelatedDropdownFilter),
97	('qc', ChoiceDropdownFilter)
98	]
99	search_fields = ('session__id', 'name', 'collection', 'dataset_type__name',	1✔
100	'dataset_type__filename_pattern', 'version')
101	ordering = ('-created_datetime',)	1✔
102
103	def get_queryset(self, request):	1✔
104	queryset = super(DatasetAdmin, self).get_queryset(request)	1✔
105	queryset = queryset.select_related('session', 'session__subject', 'created_by')	1✔
106	return queryset	1✔
107
108	def dataset_type_(self, obj):	1✔
109	return obj.dataset_type.name	×
110
111	def name_(self, obj):	1✔
112	return obj.name or '<unnamed>'	×
113
114	def revision_(self, obj):	1✔
115	return obj.revision.name	1✔
116
117	def session_ro(self, obj):	1✔
118	url = get_admin_url(obj.session)	1✔
119	return format_html('<a href="{url}">{name}</a>', url=url, name=obj.session)	1✔
120	session_ro.short_description = 'session'	1✔
121
122	def subject(self, obj):	1✔
123	return obj.session.subject.nickname	×
124
125	def _online(self, obj):	1✔
126	return obj.is_online	1✔
127	_online.short_description = 'On server'	1✔
128	_online.boolean = True	1✔
129
130	def _protected(self, obj):	1✔
131	return obj.is_protected	1✔
132	_protected.short_description = 'Protected'	1✔
133	_protected.boolean = True	1✔
134
135	def _public(self, obj):	1✔
136	return obj.is_public	1✔
137	_public.short_description = 'Public'	1✔
138	_public.boolean = True	1✔
139
140	def delete_queryset(self, request, queryset):	1✔
141	try:	×
142	queryset.delete()	×
143	except ProtectedError as e:	×
144	err_msg = e.args[0] if e.args else 'One or more dataset(s) protected'	×
145	self.message_user(request, err_msg, level=messages.ERROR)	×
146
147	def delete_model(self, request, obj):	1✔
148	try:	×
149	obj.delete()	×
150	except ProtectedError as e:	×
151	# FIXME This still shows the successful message which is confusing for users
152	err_msg = e.args[0] if e.args else f'Dataset {obj.name} is protected'	×
153	self.message_user(request, err_msg, level=messages.ERROR)	×
154
155
156	class FileRecordAdmin(BaseAdmin):	1✔
157	fields = ('relative_path', 'data_repository', 'dataset', 'exists')	1✔
158	list_display = ('relative_path', 'repository', 'dataset_name',	1✔
159	'user', 'datetime', 'exists')
160	readonly_fields = ('dataset', 'dataset_name', 'repository', 'user', 'datetime')	1✔
161	list_filter = ('exists', 'data_repository__name')	1✔
162	search_fields = ('dataset__created_by__username', 'dataset__name',	1✔
163	'relative_path', 'data_repository__name')
164	ordering = ('-dataset__created_datetime',)	1✔
165
166	def get_queryset(self, request):	1✔
167	qs = super(FileRecordAdmin, self).get_queryset(request)	1✔
168	qs = qs.select_related('data_repository', 'dataset', 'dataset__created_by')	1✔
169	return qs	1✔
170
171	def repository(self, obj):	1✔
172	return getattr(obj.data_repository, 'name', None)	×
173
174	def dataset_name(self, obj):	1✔
175	return getattr(obj.dataset, 'name', None)	×
176
177	def user(self, obj):	1✔
178	return getattr(obj.dataset, 'created_by', None)	×
179
180	def datetime(self, obj):	1✔
181	return getattr(obj.dataset, 'created_datetime', None)	×
182
183
184	class DownloadAdmin(BaseAdmin):	1✔
185	fields = ('user', 'dataset', 'first_download', 'last_download', 'count', 'projects')	1✔
186	autocomplete_fields = ('dataset',)	1✔
187	readonly_fields = ('first_download', 'last_download')	1✔
188	list_display = ('dataset_type', 'dataset_name', 'subject', 'created_by',	1✔
189	'user', 'first_download', 'last_download', 'count')
190	list_display_links = ('first_download',)	1✔
191	search_filter = ('user__username', 'dataset__name')	1✔
192
193	def dataset_name(self, obj):	1✔
194	return obj.dataset.name	×
195
196	def subject(self, obj):	1✔
197	return obj.dataset.session.subject.nickname	×
198
199	def dataset_type(self, obj):	1✔
200	return obj.dataset.dataset_type.name	×
201
202	def created_by(self, obj):	1✔
203	return obj.dataset.created_by.username	×
204
205
206	class RevisionAdmin(BaseAdmin):	1✔
207	fields = ['name', 'description', 'created_datetime']	1✔
208	readonly_fields = ['created_datetime']	1✔
209	list_display = ['name', 'description']	1✔
210	search_fields = ('name',)	1✔
211	ordering = ('-created_datetime',)	1✔
212
213
214	class TagAdmin(BaseAdmin):	1✔
215	fields = ['name', 'description', 'protected', 'public', 'dataset_count', 'session_count']	1✔
216	list_display = ['name', 'description', 'dataset_count', 'session_count', 'protected', 'public']	1✔
217	readonly_fields = ['dataset_count', 'session_count']	1✔
218	search_fields = ('name',)	1✔
219	ordering = ('name',)	1✔
220
221	def dataset_count(self, tag):	1✔
222	return tag.dataset_count	1✔
223
224	def session_count(self, tag):	1✔
225	return tag.session_count	1✔
226
227	def get_queryset(self, request):	1✔
228	queryset = super().get_queryset(request)	1✔
229	queryset = queryset.annotate(dataset_count=Count("datasets"))	1✔
230	queryset = queryset.annotate(session_count=Count("datasets__session", distinct=True))	1✔
231	return queryset	1✔
232
233
234	admin.site.register(DataRepositoryType, DataRepositoryTypeAdmin)	1✔
235	admin.site.register(DataRepository, DataRepositoryAdmin)	1✔
236	admin.site.register(DataFormat, DataFormatAdmin)	1✔
237	admin.site.register(DatasetType, DatasetTypeAdmin)	1✔
238	admin.site.register(Dataset, DatasetAdmin)	1✔
239	admin.site.register(FileRecord, FileRecordAdmin)	1✔
240	admin.site.register(Download, DownloadAdmin)	1✔
241	admin.site.register(Revision, RevisionAdmin)	1✔
242	admin.site.register(Tag, TagAdmin)	1✔

cortex-lab / alyx / 29940300361

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