15909158736

Committed 26 Jun 2025 06:03PM UTC coverage: 81.092% (-0.6%) from 81.702%

Build # 15909158736

Build Type

Pull #850

github

Committed by

web-flow

Commit Message

Merge pull request #875 from bodintsov/feature/share-cleanupgrade-2025-type-annotations

[ENG-7443] Feature/share cleanupgrade 2025 type annotations

Pull Request Pull Request #850: [project][ENG-7225] share clean(up)grade 2025 (milestone 2: upgrade)

Run Details

485 of 534 new or added lines in 63 files covered. (90.82%)

12 existing lines in 5 files now uncovered.

6150 of 7584 relevant lines covered (81.09%)

0.81 hits per line

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

59.15

/share/admin/celery.py

from functools import reduce
import operator
import pprint
import uuid

from celery import states

from django.db import models
from django.contrib import admin
from django.utils.html import format_html

from project import celery_app

from share.admin.util import TimeLimitedPaginator


class TaskNameFilter(admin.SimpleListFilter):
    title = 'Task'
    parameter_name = 'task_name'

    def lookups(self, request, model_admin):
        celery_app.autodiscover_tasks([
            'share',
            'share.janitor',
        ], force=True)
        return sorted((x, x) for x in celery_app.tasks.keys())

    def queryset(self, request, queryset):
        if self.value():
            return queryset.filter(task_name=self.value())
        return queryset


class StatusFilter(admin.SimpleListFilter):
    title = 'Status'
    parameter_name = 'status'

    def lookups(self, request, model_admin):
        return sorted((x, x.title()) for x in states.ALL_STATES)

    def queryset(self, request, queryset):
        _value = self.value()
        if _value:
            return queryset.filter(status=_value.upper())
        return queryset


class CeleryTaskResultAdmin(admin.ModelAdmin):
    list_display = ('task_id', 'task_name', 'status_', 'source_', 'date_modified', 'date_created', 'share_version')
    exclude = ('correlation_id', )
    actions = ('retry', )
    ordering = ('-date_modified', )
    list_filter = (TaskNameFilter, StatusFilter, )
    readonly_fields = (
        'task_id',
        'task_name',
        'task_args', 'task_kwargs',
        'result', 'traceback',
        'meta_',
        'date_created', 'date_modified',
        'share_version'
    )
    show_full_result_count = False
    paginator = TimeLimitedPaginator
    search_fields = ('task_name', )

    STATUS_COLORS = {
        states.SUCCESS: 'green',
        states.FAILURE: 'red',
        states.STARTED: 'cyan',
        states.RETRY: 'orange',
    }

    def get_search_results(self, request, queryset, search_term):
        try:
            return queryset.filter(task_id=uuid.UUID(search_term)), False
        except ValueError:
            pass

        # Overriden because there is no way to opt out of a case insensitive search
        search_fields = self.get_search_fields(request)
        use_distinct = bool(search_term)
        if search_fields and search_term:
            orm_lookups = ['{}__startswith'.format(search_field) for search_field in search_fields]
            for bit in search_term.split():
                or_queries = [models.Q(**{orm_lookup: bit}) for orm_lookup in orm_lookups]
                queryset = queryset.filter(reduce(operator.or_, or_queries))

        return queryset, use_distinct

    def task_args(self, obj):
        return obj.meta['args']

    def task_kwargs(self, obj):
        return pprint.pformat(obj.meta['kwargs'])

    def status_(self, obj):
        return format_html(
            '<span style="font-weight: bold; color: {}">{}</span>',
            self.STATUS_COLORS.get(obj.status, 'black'),
            obj.status.title()
        )
    status_.short_description = 'Status'  # type: ignore[attr-defined]

    def meta_(self, obj):
        return pprint.pformat(obj.meta)
    meta_.short_description = 'Meta'  # type: ignore[attr-defined]

    def source_(self, obj):
        return obj.meta.get('source_config') or obj.meta.get('source')
    source_.short_description = 'Source'  # type: ignore[attr-defined]

    def retry(self, request, queryset):
        for task in queryset:
            celery_app.tasks[task.task_name].apply_async(
                task.meta.get('args', ()),
                task.meta.get('kwargs', {}),
                task_id=str(task.task_id)
            )
    retry.short_description = 'Retry Tasks'  # type: ignore[attr-defined]

1	from functools import reduce	1✔
2	import operator	1✔
3	import pprint	1✔
4	import uuid	1✔
5
6	from celery import states	1✔
7
8	from django.db import models	1✔
9	from django.contrib import admin	1✔
10	from django.utils.html import format_html	1✔
11
12	from project import celery_app	1✔
13
14	from share.admin.util import TimeLimitedPaginator	1✔
15
16
17	class TaskNameFilter(admin.SimpleListFilter):	1✔
18	title = 'Task'	1✔
19	parameter_name = 'task_name'	1✔
20
21	def lookups(self, request, model_admin):	1✔
22	celery_app.autodiscover_tasks([	×
23	'share',
24	'share.janitor',
25	], force=True)
26	return sorted((x, x) for x in celery_app.tasks.keys())	×
27
28	def queryset(self, request, queryset):	1✔
29	if self.value():	×
30	return queryset.filter(task_name=self.value())	×
31	return queryset	×
32
33
34	class StatusFilter(admin.SimpleListFilter):	1✔
35	title = 'Status'	1✔
36	parameter_name = 'status'	1✔
37
38	def lookups(self, request, model_admin):	1✔
39	return sorted((x, x.title()) for x in states.ALL_STATES)	×
40
41	def queryset(self, request, queryset):	1✔
NEW 42	_value = self.value()	×
NEW 43	if _value:	×
NEW 44	return queryset.filter(status=_value.upper())	×
UNCOV 45	return queryset	×
46
47
48	class CeleryTaskResultAdmin(admin.ModelAdmin):	1✔
49	list_display = ('task_id', 'task_name', 'status_', 'source_', 'date_modified', 'date_created', 'share_version')	1✔
50	exclude = ('correlation_id', )	1✔
51	actions = ('retry', )	1✔
52	ordering = ('-date_modified', )	1✔
53	list_filter = (TaskNameFilter, StatusFilter, )	1✔
54	readonly_fields = (	1✔
55	'task_id',
56	'task_name',
57	'task_args', 'task_kwargs',
58	'result', 'traceback',
59	'meta_',
60	'date_created', 'date_modified',
61	'share_version'
62	)
63	show_full_result_count = False	1✔
64	paginator = TimeLimitedPaginator	1✔
65	search_fields = ('task_name', )	1✔
66
67	STATUS_COLORS = {	1✔
68	states.SUCCESS: 'green',
69	states.FAILURE: 'red',
70	states.STARTED: 'cyan',
71	states.RETRY: 'orange',
72	}
73
74	def get_search_results(self, request, queryset, search_term):	1✔
75	try:	×
76	return queryset.filter(task_id=uuid.UUID(search_term)), False	×
77	except ValueError:	×
78	pass	×
79
80	# Overriden because there is no way to opt out of a case insensitive search
81	search_fields = self.get_search_fields(request)	×
82	use_distinct = bool(search_term)	×
83	if search_fields and search_term:	×
84	orm_lookups = ['{}__startswith'.format(search_field) for search_field in search_fields]	×
85	for bit in search_term.split():	×
86	or_queries = [models.Q(**{orm_lookup: bit}) for orm_lookup in orm_lookups]	×
87	queryset = queryset.filter(reduce(operator.or_, or_queries))	×
88
89	return queryset, use_distinct	×
90
91	def task_args(self, obj):	1✔
92	return obj.meta['args']	×
93
94	def task_kwargs(self, obj):	1✔
95	return pprint.pformat(obj.meta['kwargs'])	×
96
97	def status_(self, obj):	1✔
98	return format_html(	×
99	'<span style="font-weight: bold; color: {}">{}</span>',
100	self.STATUS_COLORS.get(obj.status, 'black'),
101	obj.status.title()
102	)
103	status_.short_description = 'Status' # type: ignore[attr-defined]	1✔
104
105	def meta_(self, obj):	1✔
106	return pprint.pformat(obj.meta)	×
107	meta_.short_description = 'Meta' # type: ignore[attr-defined]	1✔
108
109	def source_(self, obj):	1✔
110	return obj.meta.get('source_config') or obj.meta.get('source')	×
111	source_.short_description = 'Source' # type: ignore[attr-defined]	1✔
112
113	def retry(self, request, queryset):	1✔
114	for task in queryset:	×
115	celery_app.tasks[task.task_name].apply_async(	×
116	task.meta.get('args', ()),
117	task.meta.get('kwargs', {}),
118	task_id=str(task.task_id)
119	)
120	retry.short_description = 'Retry Tasks' # type: ignore[attr-defined]	1✔

CenterForOpenScience / SHARE / 15909158736

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