#6402

Committed 10 Jun 2025 07:58AM UTC coverage: 74.933%. First build

Build # #6402

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Merge pull request #6644 from OCHA-DAP/feature/HDX-10666-implement-hub-backend

PR HDX-10666 notifications hub - implement backend (partially)

Run Details

8 of 42 new or added lines in 2 files covered. (19.05%)

12932 of 17258 relevant lines covered (74.93%)

0.75 hits per line

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

44.78

/ckanext-hdx_users/ckanext/hdx_users/views/dashboard.py

import logging

from flask import Blueprint

from ckan.common import current_user
from ckan.types import Context

import ckan.lib.base as base
import ckan.plugins.toolkit as tk
import ckan.model as model
from ckanext.hdx_users.controller_logic.dashboard_dataset_logic import DashboardDatasetLogic
from ckanext.hdx_users.general_token_model import ObjectType, get_by_type_and_user_id_and_object, TokenType
from ckanext.hdx_users.views.notification_platform import _generate_url_for

log = logging.getLogger(__name__)

render = tk.render
get_action = tk.get_action
request = tk.request
g = tk.g
h = tk.h
_ = tk._
abort = base.abort
NotFound = tk.ObjectNotFound
NotAuthorized = tk.NotAuthorized

hdx_user_dashboard = Blueprint(u'hdx_user_dashboard', __name__, url_prefix=u'/dashboard')


def datasets():
    """
    Dashboard tab for datasets. Modified to add the ability to change
    the order and ultimately filter datasets displayed
    """

    if not g.user:
        h.flash_error(_(u'Not authorized to see this page'))
        return h.redirect_to(u'home.index')

    dashboard_dataset_logic = DashboardDatasetLogic(g.userobj).read()
    if dashboard_dataset_logic.redirect_result:
        return dashboard_dataset_logic.redirect_result
    else:
        return render('user/dashboard_datasets.html', extra_vars={
            'user_dict': dashboard_dataset_logic.user_dict,
            'search_data': dashboard_dataset_logic.search_data
        })


def notifications():
    """
    Notifications tab
    """

    if not current_user.is_authenticated:
        h.flash_error(_(u'Not authorized to see this page'))
        return h.redirect_to(u'home.index')

    context: Context = {'model': model, 'session': model.Session, 'user': current_user.name}
    data_dict = {
        'user_id': current_user.id,
    }
    subscriptions = tk.get_action('hdx_notifications_subscription_list')(context, data_dict)
    for subscription in subscriptions:
        object_type = ObjectType(subscription.get('object_type'))
        object_id = subscription.get('object')
        try:
            object_dict = _get_object_dict(object_type, object_id)
            unsubscribe_token = get_by_type_and_user_id_and_object(TokenType.UNSUBSCRIBE_FOR_NOTIFICATION, current_user.email, object_type, object_id)
            subscription['object_dict'] = object_dict
            subscription['object_link'] = _generate_url_for(object_type, object_id, True)
            subscription['unsubscribe_token'] = unsubscribe_token.token if unsubscribe_token else None
        except tk.ObjectNotFound:
            raise tk.ValidationError(f'{object_type.value} {object_id} does not exist')
        except Exception as e:
            log.error(f'Error retrieving target or user: {e}')
            raise e

    return render('user/dashboard_notifications.html', extra_vars={
        'subscriptions': subscriptions,
        'user_dict': current_user,
    })

def _get_object_dict(object_type, object_id):
    context: Context = {}

    action = None
    if object_type == ObjectType.DATASET.value:
        action = 'package_show'
    elif object_type == ObjectType.GROUP.value:
        action = 'group_show'
    elif object_type == ObjectType.ORGANIZATION.value:
        action = 'organization_show'
    elif object_type == ObjectType.CRISIS.value:
        action = 'page_show'

    if action:
        return tk.get_action(action)(context, {'id': object_id})
    else:
        return {}


hdx_user_dashboard.add_url_rule(u'/datasets', view_func=datasets)
hdx_user_dashboard.add_url_rule(u'/notifications', view_func=notifications)

1	import logging	1✔
2
3	from flask import Blueprint	1✔
4
5	from ckan.common import current_user	1✔
6	from ckan.types import Context	1✔
7
8	import ckan.lib.base as base	1✔
9	import ckan.plugins.toolkit as tk	1✔
10	import ckan.model as model	1✔
11	from ckanext.hdx_users.controller_logic.dashboard_dataset_logic import DashboardDatasetLogic	1✔
12	from ckanext.hdx_users.general_token_model import ObjectType, get_by_type_and_user_id_and_object, TokenType	1✔
13	from ckanext.hdx_users.views.notification_platform import _generate_url_for	1✔
14
15	log = logging.getLogger(__name__)	1✔
16
17	render = tk.render	1✔
18	get_action = tk.get_action	1✔
19	request = tk.request	1✔
20	g = tk.g	1✔
21	h = tk.h	1✔
22	_ = tk._	1✔
23	abort = base.abort	1✔
24	NotFound = tk.ObjectNotFound	1✔
25	NotAuthorized = tk.NotAuthorized	1✔
26
27	hdx_user_dashboard = Blueprint(u'hdx_user_dashboard', __name__, url_prefix=u'/dashboard')	1✔
28
29
30	def datasets():	1✔
31	"""
32	Dashboard tab for datasets. Modified to add the ability to change
33	the order and ultimately filter datasets displayed
34	"""
35
36	if not g.user:	1✔
37	h.flash_error(_(u'Not authorized to see this page'))	×
38	return h.redirect_to(u'home.index')	×
39
40	dashboard_dataset_logic = DashboardDatasetLogic(g.userobj).read()	1✔
41	if dashboard_dataset_logic.redirect_result:	1✔
42	return dashboard_dataset_logic.redirect_result	×
43	else:
44	return render('user/dashboard_datasets.html', extra_vars={	1✔
45	'user_dict': dashboard_dataset_logic.user_dict,
46	'search_data': dashboard_dataset_logic.search_data
47	})
48
49
50	def notifications():	1✔
51	"""
52	Notifications tab
53	"""
54
55	if not current_user.is_authenticated:	×
56	h.flash_error(_(u'Not authorized to see this page'))	×
57	return h.redirect_to(u'home.index')	×
58
NEW 59	context: Context = {'model': model, 'session': model.Session, 'user': current_user.name}	×
NEW 60	data_dict = {	×
61	'user_id': current_user.id,
62	}
NEW 63	subscriptions = tk.get_action('hdx_notifications_subscription_list')(context, data_dict)	×
NEW 64	for subscription in subscriptions:	×
NEW 65	object_type = ObjectType(subscription.get('object_type'))	×
NEW 66	object_id = subscription.get('object')	×
NEW 67	try:	×
NEW 68	object_dict = _get_object_dict(object_type, object_id)	×
NEW 69	unsubscribe_token = get_by_type_and_user_id_and_object(TokenType.UNSUBSCRIBE_FOR_NOTIFICATION, current_user.email, object_type, object_id)	×
NEW 70	subscription['object_dict'] = object_dict	×
NEW 71	subscription['object_link'] = _generate_url_for(object_type, object_id, True)	×
NEW 72	subscription['unsubscribe_token'] = unsubscribe_token.token if unsubscribe_token else None	×
NEW 73	except tk.ObjectNotFound:	×
NEW 74	raise tk.ValidationError(f'{object_type.value} {object_id} does not exist')	×
NEW 75	except Exception as e:	×
NEW 76	log.error(f'Error retrieving target or user: {e}')	×
NEW 77	raise e	×
78
79	return render('user/dashboard_notifications.html', extra_vars={	×
80	'subscriptions': subscriptions,
81	'user_dict': current_user,
82	})
83
84	def _get_object_dict(object_type, object_id):	1✔
NEW 85	context: Context = {}	×
86
NEW 87	action = None	×
NEW 88	if object_type == ObjectType.DATASET.value:	×
NEW 89	action = 'package_show'	×
NEW 90	elif object_type == ObjectType.GROUP.value:	×
NEW 91	action = 'group_show'	×
NEW 92	elif object_type == ObjectType.ORGANIZATION.value:	×
NEW 93	action = 'organization_show'	×
NEW 94	elif object_type == ObjectType.CRISIS.value:	×
NEW 95	action = 'page_show'	×
96
NEW 97	if action:	×
NEW 98	return tk.get_action(action)(context, {'id': object_id})	×
99	else:
NEW 100	return {}	×
101
102
103	hdx_user_dashboard.add_url_rule(u'/datasets', view_func=datasets)	1✔
104	hdx_user_dashboard.add_url_rule(u'/notifications', view_func=notifications)	1✔

OCHA-DAP / hdx-ckan / #6402

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