mypy changes

This commit is contained in:
mike cullerton 2022-06-29 14:45:29 -04:00
parent 6658cc9140
commit 25201a6045
3 changed files with 11 additions and 6 deletions

View File

@ -1,6 +1,7 @@
"""Task."""
import enum
from typing import Any
from typing import Optional
import marshmallow
from marshmallow import Schema
@ -91,7 +92,7 @@ class Task:
type: str,
state: str,
lane: str,
form: str,
form: Optional[dict],
documentation: str,
data: dict[str, Any],
multi_instance_type: str,

View File

@ -12,6 +12,7 @@ from flask import g
from flask import Response
from flask_bpmn.api.api_error import ApiError
from flask_bpmn.models.db import db
from sqlalchemy import desc
from spiffworkflow_backend.exceptions.process_entity_not_found_error import (
ProcessEntityNotFoundError,
@ -446,7 +447,7 @@ def task_list_my_tasks(page: int = 1, per_page: int = 100) -> flask.wrappers.Res
active_tasks = (
ActiveTaskModel.query.filter_by(assigned_principal_id=principal.id)
.order_by(ActiveTaskModel.id.desc())
.order_by(desc(ActiveTaskModel.id))
.paginate(page, per_page, False)
)

View File

@ -14,6 +14,7 @@ from SpiffWorkflow.bpmn.specs.events import StartEvent
from SpiffWorkflow.bpmn.specs.ManualTask import ManualTask
from SpiffWorkflow.bpmn.specs.ScriptTask import ScriptTask
from SpiffWorkflow.bpmn.specs.UserTask import UserTask
from SpiffWorkflow.camunda.specs.UserTask import EnumFormField
from SpiffWorkflow.dmn.specs.BusinessRuleTask import BusinessRuleTask
from SpiffWorkflow.specs import CancelTask
from SpiffWorkflow.specs import StartTask
@ -169,7 +170,7 @@ class ProcessInstanceService:
return result
@staticmethod
def update_task_assignments(processor):
def update_task_assignments(processor: ProcessInstanceProcessor) -> None:
"""For every upcoming user task, log a task action that connects the assigned user(s) to that task.
All existing assignment actions for this workflow are removed from the database,
@ -212,7 +213,9 @@ class ProcessInstanceService:
):
current_user = spiff_task.data["current_user"]
principal = UserService().get_principal_by_user_id(current_user.id)
return (principal.id,)
return [
principal.id,
]
# return [processor.process_instance_model.process_initiator_id]
if spiff_task.task_spec.lane not in spiff_task.data:
@ -514,7 +517,7 @@ class ProcessInstanceService:
return field
@staticmethod
def get_options_from_task_data(spiff_task, field):
def get_options_from_task_data(spiff_task: Task, field: EnumFormField) -> List:
"""Get_options_from_task_data."""
prop = field.get_property(Task.FIELD_PROP_DATA_NAME)
if prop not in spiff_task.data:
@ -529,7 +532,7 @@ class ProcessInstanceService:
value_column = field.get_property(Task.FIELD_PROP_VALUE_COLUMN)
label_column = field.get_property(Task.FIELD_PROP_LABEL_COLUMN)
items = data_model.items() if isinstance(data_model, dict) else data_model
options = []
options: List[Any] = []
for item in items:
if value_column not in item:
raise ApiError.from_task(