Skip to content

Services

hypha.apply.todo.services

validate_user_uniqueness 

validate_user_uniqueness(code, user, related_obj)

code + related_object + user should be unique together.

Source code in hypha/apply/todo/services.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def validate_user_uniqueness(code, user, related_obj):
    """
    code + related_object + user should be unique together.
    """
    matching_tasks = Task.objects.filter(
        code=code,
        related_content_type=ContentType.objects.get_for_model(related_obj).id,
        related_object_id=related_obj.id,
    )
    if matching_tasks.filter(user=user).exists():
        # if same task already assigned to the same user
        # raise ValidationError("Task is already assigned to the user") # :todo: add validation msg as a log msg?
        return False
    else:
        # if same task is already assigned to user's user_group
        user_group_matching_tasks = matching_tasks.annotate(
            group_count=Count("user_group")
        ).filter(group_count=len(user.groups.all()))
        for group in user.groups.all():
            user_group_matching_tasks = user_group_matching_tasks.filter(
                user_group__id=group.id
            )
        if user_group_matching_tasks.exists():
            # raise ValidationError("Task is already assigned to user's group")
            return False
        return True

validate_user_groups_uniqueness 

validate_user_groups_uniqueness(code, user_groups, related_obj)

code + related_object + user_group should be unique together.

Source code in hypha/apply/todo/services.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def validate_user_groups_uniqueness(code, user_groups, related_obj):
    """
    code + related_object + user_group should be unique together.
    """
    matching_tasks = Task.objects.filter(
        code=code,
        related_content_type=ContentType.objects.get_for_model(related_obj).id,
        related_object_id=related_obj.id,
    )
    user_group_matching_tasks = matching_tasks.annotate(
        group_count=Count("user_group")
    ).filter(group_count=len(user_groups))
    for group in user_groups:
        user_group_matching_tasks = user_group_matching_tasks.filter(
            user_group__id=group.id
        )
    if user_group_matching_tasks.exists():
        # same task with same user_group already exists
        # :todo: add validation msg as a log msg?
        return False

    # user with exact user group already assigned for same task
    users = get_users_for_groups(
        list(user_groups), exact_match=True
    )  # users with provided user_group

    for user in users:
        if matching_tasks.filter(user=user).exists():
            Task.objects.filter(
                id=matching_tasks.id
            ).delete()  # delete those user's tasks
    return True