Feature: manual workflow trigger (#14704)

This commit is contained in:
shamoon authored and GitHub committed 2026-11-12 13:28:10 -08:00
1 parent 62d729e301
commit 3dbeca603e
27 files changed
+825 -8

No files matched your search

+20
View File
@@ -42,6 +42,7 @@ from documents.plugins.helpers import DocumentsStatusManager
from documents.tasks import bulk_update_documents
from documents.tasks import consume_file
from documents.tasks import remove_document_from_index
from documents.tasks import run_workflows_for_documents
from documents.tasks import update_document_content_maybe_archive_file
from documents.versioning import get_latest_version_for_root
from documents.versioning import get_root_document
@@ -464,6 +465,25 @@ def reprocess(doc_ids: list[int], *, remote_ocr: bool = False) -> Literal["OK"]:
return "OK"
def run_workflow(
doc_ids: list[int],
workflow: int,
*,
filter_documents: bool = True,
) -> Literal["OK"]:
"""
Manually run a single workflow against the given documents.
Done asynchronously via a Celery task.
"""
run_workflows_for_documents.delay(
document_ids=doc_ids,
workflow_id=workflow,
filter_documents=filter_documents,
)
return "OK"
def set_permissions(
doc_ids: list[int],
set_permissions: PermissionsSpec,
+13
View File
@@ -35,6 +35,7 @@ from django_filters.rest_framework import DateTimeFilter
from django_filters.rest_framework import Filter
from django_filters.rest_framework import FilterSet
from django_filters.rest_framework import MultipleChoiceFilter
from django_filters.rest_framework import NumberFilter
from drf_spectacular.utils import extend_schema_field
from guardian.utils import get_group_obj_perms_model
from guardian.utils import get_user_obj_perms_model
@@ -52,6 +53,7 @@ from documents.models import ShareLink
from documents.models import ShareLinkBundle
from documents.models import StoragePath
from documents.models import Tag
from documents.models import Workflow
from documents.permissions import annotate_authorizing_fields
from documents.permissions import permitted_document_ids
from documents.permissions import permitted_object_ids
@@ -249,6 +251,17 @@ class CustomFieldFilterSet(FilterSet):
}
class WorkflowFilterSet(FilterSet):
trigger_type = NumberFilter(field_name="triggers__type", distinct=True)
class Meta:
model = Workflow
fields = {
"id": ID_KWARGS,
"enabled": ["exact"],
}
@extend_schema_field(serializers.CharField)
class CustomFieldsFilter(Filter):
def filter(self, qs, value):
+1
View File
@@ -684,6 +684,7 @@ def document_matches_workflow(
trigger_type == WorkflowTrigger.WorkflowTriggerType.DOCUMENT_ADDED
or trigger_type == WorkflowTrigger.WorkflowTriggerType.DOCUMENT_UPDATED
or trigger_type == WorkflowTrigger.WorkflowTriggerType.SCHEDULED
or trigger_type == WorkflowTrigger.WorkflowTriggerType.MANUAL
):
trigger_matched, reason = existing_document_matches_workflow(
document,
@@ -0,0 +1,43 @@
# Generated by Django 5.2.18 on 2026-11-06 17:12
from django.db import migrations
from django.db import models
class Migration(migrations.Migration):
dependencies = [
("documents", "0028_replace_tag_tree_fields"),
]
operations = [
migrations.AlterField(
model_name="workflowrun",
name="type",
field=models.PositiveSmallIntegerField(
choices=[
(1, "Consumption Started"),
(2, "Document Added"),
(3, "Document Updated"),
(4, "Scheduled"),
(5, "Manual"),
],
null=True,
verbose_name="workflow trigger type",
),
),
migrations.AlterField(
model_name="workflowtrigger",
name="type",
field=models.PositiveSmallIntegerField(
choices=[
(1, "Consumption Started"),
(2, "Document Added"),
(3, "Document Updated"),
(4, "Scheduled"),
(5, "Manual"),
],
default=1,
verbose_name="Workflow Trigger Type",
),
),
]
+1
View File
@@ -1532,6 +1532,7 @@ class WorkflowTrigger(models.Model):
DOCUMENT_ADDED = 2, _("Document Added")
DOCUMENT_UPDATED = 3, _("Document Updated")
SCHEDULED = 4, _("Scheduled")
MANUAL = 5, _("Manual")
class DocumentSourceChoices(models.IntegerChoices):
CONSUME_FOLDER = DocumentSource.ConsumeFolder.value, _("Consume Folder")
+34
View File
@@ -1829,6 +1829,7 @@ class BulkEditSerializer(
"modify_tags",
"modify_custom_fields",
"set_permissions",
"run_workflow",
*LEGACY_DOCUMENT_ACTION_METHODS,
],
label="Method",
@@ -1918,6 +1919,8 @@ class BulkEditSerializer(
return bulk_edit.reprocess
elif method == "set_permissions":
return bulk_edit.set_permissions
elif method == "run_workflow":
return bulk_edit.run_workflow
elif method == "rotate":
return bulk_edit.rotate
elif method == "merge":
@@ -2157,6 +2160,35 @@ class BulkEditSerializer(
else:
parameters["remote_ocr"] = False
def _validate_parameters_run_workflow(self, parameters) -> None:
if "workflow" not in parameters:
raise serializers.ValidationError("workflow not specified")
try:
workflow = Workflow.objects.get(id=parameters["workflow"])
except (Workflow.DoesNotExist, ValueError, TypeError):
raise serializers.ValidationError("Workflow does not exist")
request = self.context.get("request")
user = getattr(request, "user", None)
if user is not None and not user.has_perm("documents.view_workflow"):
raise PermissionDenied(
"Insufficient permissions to run this workflow",
)
if not workflow.enabled:
raise serializers.ValidationError("Workflow is not enabled")
if not workflow.triggers.filter(
type=WorkflowTrigger.WorkflowTriggerType.MANUAL,
).exists():
raise serializers.ValidationError(
"Workflow does not have a manual trigger",
)
if "filter_documents" in parameters:
if not isinstance(parameters["filter_documents"], bool):
raise serializers.ValidationError(
"filter_documents must be a boolean",
)
else:
parameters["filter_documents"] = True
def validate_parameters_remove_password(self, parameters):
if "password" not in parameters:
raise serializers.ValidationError("password not specified")
@@ -2223,6 +2255,8 @@ class BulkEditSerializer(
self.validate_parameters_remove_password(parameters)
elif method == bulk_edit.reprocess:
self._validate_parameters_reprocess(parameters)
elif method == bulk_edit.run_workflow:
self._validate_parameters_run_workflow(parameters)
return attrs
+11 -1
View File
@@ -864,6 +864,8 @@ def run_workflows(
logging_group: uuid.UUID | None = None,
overrides: DocumentMetadataOverrides | None = None,
original_file: Path | None = None,
*,
skip_matching: bool = False,
) -> tuple[DocumentMetadataOverrides, str] | None:
"""
Execute workflows matching a document for the given trigger. When `overrides` is provided
@@ -874,6 +876,10 @@ def run_workflows(
`document.source_path` (Document) or `document.original_file` (ConsumableDocument).
Passing `workflow_to_run` skips the workflow query (currently only used by scheduled runs).
Passing `skip_matching` runs the workflow's actions without evaluating its trigger
filters against the document. It is only meaningful together with `workflow_to_run`
(e.g. a manual run the user has explicitly forced onto the selected documents).
"""
use_overrides = overrides is not None
@@ -923,7 +929,11 @@ def run_workflows(
)
break
if matching.document_matches_workflow(document, workflow, trigger_type):
if skip_matching or matching.document_matches_workflow(
document,
workflow,
trigger_type,
):
action: WorkflowAction
has_move_to_trash_action = False
for action in workflow.actions.order_by("order", "pk"):
+46
View File
@@ -53,6 +53,7 @@ from documents.models import ShareLink
from documents.models import ShareLinkBundle
from documents.models import StoragePath
from documents.models import Tag
from documents.models import Workflow
from documents.models import WorkflowRun
from documents.models import WorkflowTrigger
from documents.plugins.base import ConsumeTaskPlugin
@@ -691,6 +692,51 @@ def check_scheduled_workflows() -> None:
)
@shared_task
def run_workflows_for_documents(
document_ids: list[int],
workflow_id: int,
*,
filter_documents: bool = True,
) -> None:
"""
Run a single workflow manually against the given documents.
``filter_documents`` controls whether the workflow's trigger filters are applied.
When True, only documents matching the existing filters will have the workflow run.
When False, the workflow will be run on all specified documents regardless of the filters.
``
"""
try:
workflow = Workflow.objects.get(id=workflow_id)
except Workflow.DoesNotExist:
logger.warning(
f"Manual workflow run requested for missing workflow {workflow_id}",
)
return
for document in Document.objects.filter(
id__in=document_ids,
root_document__isnull=True,
):
try:
run_workflows(
trigger_type=WorkflowTrigger.WorkflowTriggerType.MANUAL,
document=document,
workflow_to_run=workflow,
skip_matching=not filter_documents,
)
# Manual runs dont send document_updated signal, so ws here to ensure clients are refreshed
send_websocket_document_updated(
sender=None,
document=document,
)
except Exception:
logger.exception(
f"Error running workflow {workflow} on document {document.pk}",
)
def update_document_parent_tags(tag: Tag, new_parent: Tag) -> None:
"""
When a tag's parent changes, ensure all documents containing the tag also have
+109
View File
@@ -15,7 +15,11 @@ from documents.models import Document
from documents.models import DocumentType
from documents.models import StoragePath
from documents.models import Tag
from documents.models import Workflow
from documents.models import WorkflowAction
from documents.models import WorkflowTrigger
from paperless_testing.dirs import DirectoriesMixin
from paperless_testing.factories import DocumentFactory
from paperless_testing.factories import UserFactory
from paperless_testing.permissions import grant_all_global
from paperless_testing.permissions import grant_global
@@ -622,6 +626,111 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
self.assertEqual(args[0], [self.doc1.id])
self.assertEqual(kwargs, {"remote_ocr": True})
def _make_manual_workflow(self) -> Workflow:
trigger = WorkflowTrigger.objects.create(
type=WorkflowTrigger.WorkflowTriggerType.MANUAL,
)
action = WorkflowAction.objects.create()
workflow = Workflow.objects.create(name="Manual workflow", order=0)
workflow.triggers.add(trigger)
workflow.actions.add(action)
return workflow
def _run_workflow(self, parameters, documents=None):
return self.client.post(
"/api/documents/bulk_edit/",
json.dumps(
{
"documents": documents or [self.doc1.id],
"method": "run_workflow",
"parameters": parameters,
},
),
content_type="application/json",
)
@mock.patch("documents.views.bulk_edit.run_workflow")
def test_run_workflow(self, m) -> None:
"""A valid run_workflow call passes the docs, workflow id and filter flag through."""
self.setup_mock(m, "run_workflow")
workflow = self._make_manual_workflow()
response = self._run_workflow({"workflow": workflow.id})
self.assertEqual(response.status_code, status.HTTP_200_OK)
args, kwargs = m.call_args
self.assertEqual(args[0], [self.doc1.id])
self.assertEqual(kwargs, {"workflow": workflow.id, "filter_documents": True})
self._run_workflow({"workflow": workflow.id, "filter_documents": False})
self.assertFalse(m.call_args.kwargs["filter_documents"])
@mock.patch("documents.views.bulk_edit.run_workflow")
def test_run_workflow_invalid_parameters(self, m) -> None:
"""Invalid run_workflow parameters are rejected before dispatch."""
self.setup_mock(m, "run_workflow")
manual = self._make_manual_workflow()
non_manual = Workflow.objects.create(name="Non-manual", order=0)
non_manual.triggers.add(
WorkflowTrigger.objects.create(
type=WorkflowTrigger.WorkflowTriggerType.DOCUMENT_ADDED,
),
)
disabled = Workflow.objects.create(
name="Disabled manual",
order=0,
enabled=False,
)
disabled.triggers.add(
WorkflowTrigger.objects.create(
type=WorkflowTrigger.WorkflowTriggerType.MANUAL,
),
)
for label, parameters in (
("missing workflow", {}),
("nonexistent workflow", {"workflow": 9999}),
("no manual trigger", {"workflow": non_manual.id}),
("disabled workflow", {"workflow": disabled.id}),
("non-bool filter", {"workflow": manual.id, "filter_documents": "nope"}),
):
with self.subTest(label):
response = self._run_workflow(parameters)
self.assertEqual(
response.status_code,
status.HTTP_400_BAD_REQUEST,
)
m.assert_not_called()
@mock.patch("documents.views.bulk_edit.run_workflow")
def test_run_workflow_requires_ownership(self, m) -> None:
"""A non-owner without document ownership is forbidden (actions mutate docs)."""
self.setup_mock(m, "run_workflow")
user = UserFactory(username="no_perms")
grant_all_global(user)
doc = DocumentFactory(owner=UserFactory(username="other"))
workflow = self._make_manual_workflow()
self.client.force_authenticate(user=user)
response = self._run_workflow({"workflow": workflow.id}, documents=[doc.id])
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
m.assert_not_called()
@mock.patch("documents.views.bulk_edit.run_workflow")
def test_run_workflow_requires_workflow_permission(self, m) -> None:
"""Owning the documents is not enough without view_workflow permission."""
self.setup_mock(m, "run_workflow")
user = UserFactory(username="no_wf_perms")
grant_global(user, "change_document")
doc = DocumentFactory(owner=user)
workflow = self._make_manual_workflow()
self.client.force_authenticate(user=user)
response = self._run_workflow({"workflow": workflow.id}, documents=[doc.id])
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
m.assert_not_called()
@mock.patch("documents.serialisers.bulk_edit.set_storage_path")
def test_api_set_storage_path(self, m) -> None:
"""
+41
View File
@@ -100,6 +100,47 @@ class TestApiWorkflows(DirectoriesMixin, APITestCase):
self.action.assign_correspondent.pk,
)
def test_api_get_workflows_filter_trigger_type(self) -> None:
"""
GIVEN:
- A consumption workflow and a manual workflow
WHEN:
- Workflows are requested filtered by trigger_type
THEN:
- Only workflows with a matching trigger are returned
"""
manual_trigger = WorkflowTrigger.objects.create(
type=WorkflowTrigger.WorkflowTriggerType.MANUAL,
)
manual_workflow = Workflow.objects.create(name="Manual", order=1)
manual_workflow.triggers.add(manual_trigger)
response = self.client.get(
f"{self.ENDPOINT}?trigger_type={WorkflowTrigger.WorkflowTriggerType.MANUAL}",
format="json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data["count"], 1)
self.assertEqual(response.data["results"][0]["id"], manual_workflow.id)
def test_api_get_workflows_filter_enabled(self) -> None:
"""
GIVEN:
- An enabled workflow and a disabled workflow
WHEN:
- Workflows are requested filtered by enabled
THEN:
- Only enabled workflows are returned
"""
Workflow.objects.create(name="Disabled", order=1, enabled=False)
response = self.client.get(f"{self.ENDPOINT}?enabled=true", format="json")
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data["count"], 1)
self.assertEqual(response.data["results"][0]["id"], self.workflow.id)
def test_api_get_workflow_actions_ordered(self) -> None:
"""
GIVEN:
+77
View File
@@ -2288,6 +2288,83 @@ class TestWorkflows(
doc.refresh_from_db()
self.assertEqual(doc.owner, self.user2)
def _make_manual_workflow(self, *, require_tag: Tag | None = None) -> Workflow:
trigger = WorkflowTrigger.objects.create(
type=WorkflowTrigger.WorkflowTriggerType.MANUAL,
)
if require_tag is not None:
trigger.filter_has_all_tags.add(require_tag)
action = WorkflowAction.objects.create(assign_owner=self.user2)
workflow = Workflow.objects.create(name="Manual workflow", order=0)
workflow.triggers.add(trigger)
workflow.actions.add(action)
return workflow
@mock.patch("documents.tasks.send_websocket_document_updated")
def test_manual_workflow_run_applies_and_notifies(self, mock_ws) -> None:
"""A manual run applies the workflow's actions and pushes a websocket update."""
workflow = self._make_manual_workflow()
doc = DocumentFactory()
tasks.run_workflows_for_documents([doc.pk], workflow.pk)
doc.refresh_from_db()
self.assertEqual(doc.owner, self.user2)
self.assertEqual(mock_ws.call_args.kwargs["document"].pk, doc.pk)
def test_manual_workflow_run_respects_filters(self) -> None:
"""With filter_documents=True, a document failing the trigger filters is skipped."""
workflow = self._make_manual_workflow(require_tag=self.t1)
doc = DocumentFactory()
tasks.run_workflows_for_documents([doc.pk], workflow.pk, filter_documents=True)
doc.refresh_from_db()
self.assertIsNone(doc.owner)
def test_manual_workflow_run_force_ignores_filters(self) -> None:
"""With filter_documents=False, actions run regardless of the trigger filters."""
workflow = self._make_manual_workflow(require_tag=self.t1)
doc = DocumentFactory()
tasks.run_workflows_for_documents([doc.pk], workflow.pk, filter_documents=False)
doc.refresh_from_db()
self.assertEqual(doc.owner, self.user2)
def test_manual_workflow_run_missing_workflow(self) -> None:
"""A missing workflow id logs a warning rather than raising."""
doc = DocumentFactory()
with self.assertLogs("paperless.tasks", level="WARNING"):
tasks.run_workflows_for_documents([doc.pk], 9999)
@mock.patch("documents.tasks.send_websocket_document_updated")
@mock.patch("documents.tasks.run_workflows")
def test_manual_workflow_run_continues_after_error(
self,
mock_run_workflows,
mock_ws,
) -> None:
"""A failure on one document does not abort the run for the others."""
workflow = self._make_manual_workflow()
doc1 = DocumentFactory()
doc2 = DocumentFactory()
def side_effect(*args, document, **kwargs):
if document.pk == doc1.pk:
raise Exception("boom")
mock_run_workflows.side_effect = side_effect
with self.assertLogs("paperless.tasks", level="ERROR"):
tasks.run_workflows_for_documents([doc1.pk, doc2.pk], workflow.pk)
# Both documents were attempted; the websocket update only fired for the
# one that did not raise.
self.assertEqual(mock_run_workflows.call_count, 2)
ws_doc_ids = [c.kwargs["document"].pk for c in mock_ws.call_args_list]
self.assertEqual(ws_doc_ids, [doc2.pk])
@mock.patch("documents.tasks.send_websocket_document_updated")
def test_workflow_scheduled_trigger_sends_websocket_update(
self,
+98
View File
@@ -146,6 +146,7 @@ from documents.filters import ShareLinkFilterSet
from documents.filters import StoragePathFilterSet
from documents.filters import TagFilterSet
from documents.filters import TitleContentFilter
from documents.filters import WorkflowFilterSet
from documents.mail import EmailAttachment
from documents.mail import send_email
from documents.matching import match_correspondents
@@ -3008,6 +3009,100 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin):
)
return Response({"result": "OK", "task_id": task.id})
def _has_document_permissions(
self,
*,
user: User,
documents: list[int],
method,
parameters: dict[str, Any],
) -> bool:
if user.is_superuser:
return True
root_docs = {
get_root_document(doc)
for doc in Document.objects.select_related(
"owner",
"root_document__owner",
).filter(pk__in=documents)
}
user_is_owner_of_all_documents = all(
(doc.owner == user or doc.owner is None) for doc in root_docs
)
# check global and object permissions for all documents
has_perms = (
user.has_perm(
"documents.change_document",
)
and not Document.global_objects.filter(pk__in=documents)
.exclude(pk__in=permitted_document_ids(user, perm="change_document"))
.exists()
)
# check ownership for methods that change original document
if (
(
has_perms
and method
in [
bulk_edit.set_permissions,
bulk_edit.delete,
bulk_edit.rotate,
bulk_edit.delete_pages,
bulk_edit.edit_pdf,
bulk_edit.merge_as_versions,
bulk_edit.remove_password,
bulk_edit.run_workflow,
]
)
or (
method in [bulk_edit.merge, bulk_edit.split]
and parameters.get("delete_originals")
)
or (method == bulk_edit.edit_pdf and parameters.get("update_document"))
):
has_perms = has_perms and user_is_owner_of_all_documents
# check global add permissions for methods that create documents
if (
has_perms
and (
method in [bulk_edit.split, bulk_edit.merge]
or (
method in [bulk_edit.edit_pdf, bulk_edit.remove_password]
and not parameters.get("update_document")
)
)
and not user.has_perm("documents.add_document")
):
has_perms = False
# check global delete permissions for methods that delete documents
if (
has_perms
and (
method == bulk_edit.delete
# Sources stop being documents of their own, and removing one
# again afterwards needs delete_document
or method == bulk_edit.merge_as_versions
or (
method in [bulk_edit.merge, bulk_edit.split]
and parameters.get("delete_originals")
)
or (
method in [bulk_edit.edit_pdf, bulk_edit.remove_password]
and parameters.get("delete_original")
and not parameters.get("update_document")
)
)
and not user.has_perm("documents.delete_document")
):
has_perms = False
return has_perms
def _execute_document_action(
self,
*,
@@ -3084,6 +3179,7 @@ class BulkEditView(DocumentOperationPermissionMixin):
LEGACY_MODIFIED_FIELDS = {
"delete": "deleted_at",
"reprocess": "checksum",
"remove_password": None,
}
serializer_class = BulkEditSerializer
@@ -5089,6 +5185,8 @@ class WorkflowViewSet(ModelViewSet[Workflow]):
serializer_class = WorkflowSerializer
pagination_class = StandardPagination
filter_backends = (DjangoFilterBackend,)
filterset_class = WorkflowFilterSet
model = Workflow