(self)
| 644 | workspace_id = serializers.CharField(required=True, label=_('workspace id')) |
| 645 | |
| 646 | def get_query_set(self): |
| 647 | trigger_query_set = QuerySet( |
| 648 | model=get_dynamics_model({ |
| 649 | 't.name': models.CharField(), |
| 650 | 'trigger_type': models.CharField(), |
| 651 | 't.workspace_id': models.CharField(), |
| 652 | 't.is_active': models.BooleanField(), |
| 653 | 't.user_id': models.CharField(), |
| 654 | })) |
| 655 | task_query_set = QuerySet(model=get_dynamics_model({ |
| 656 | 'trigger_task_str': models.CharField(), |
| 657 | })) |
| 658 | trigger_query_set = trigger_query_set.filter(**{'t.workspace_id': self.data.get("workspace_id")}) |
| 659 | if self.data.get("name"): |
| 660 | trigger_query_set = trigger_query_set.filter(**{'t.name__icontains': self.data.get("name")}) |
| 661 | if self.data.get("type"): |
| 662 | trigger_query_set = trigger_query_set.filter(trigger_type=self.data.get("type")) |
| 663 | if self.data.get("is_active") is not None: |
| 664 | trigger_query_set = trigger_query_set.filter(**{"t.is_active": self.data.get("is_active")}) |
| 665 | if self.data.get("task"): |
| 666 | task_query_set = task_query_set.filter(trigger_task_str__icontains=self.data.get("task")) |
| 667 | if self.data.get("create_user"): |
| 668 | trigger_query_set = trigger_query_set.filter(**{"t.user_id": self.data.get("create_user")}) |
| 669 | |
| 670 | return {"trigger_query_set": trigger_query_set, "task_query_set": task_query_set} |
| 671 | |
| 672 | def page(self, current_page: int, page_size: int, with_valid=True): |
| 673 | if with_valid: |
no test coverage detected