gateway/modules/workflows/methods/methodRedmine/actions/createTicket.py
2026-04-21 21:30:11 +02:00

65 lines
2.4 KiB
Python

# Copyright (c) 2026 Patrick Motsch
# All rights reserved.
"""Workflow action: create a new Redmine ticket."""
import logging
from typing import Any, Dict
from modules.datamodels.datamodelChat import ActionResult
from modules.features.redmine.datamodelRedmine import RedmineTicketCreateRequest
from modules.features.redmine.serviceRedmine import createTicket
from ._shared import resolveInstanceContext, ticketToDict
logger = logging.getLogger(__name__)
async def createTicketAction(self, parameters: Dict[str, Any]) -> ActionResult:
"""Create a Redmine ticket. ``subject`` and ``trackerId`` are required."""
try:
user, mandateId, featureInstanceId = resolveInstanceContext(self.services, parameters)
except ValueError as exc:
return ActionResult.isFailure(error=str(exc))
subject = parameters.get("subject")
trackerId = parameters.get("trackerId")
if not subject:
return ActionResult.isFailure(error="subject is required")
try:
trackerId_int = int(trackerId) if trackerId is not None else None
except (TypeError, ValueError):
return ActionResult.isFailure(error=f"trackerId must be an int, got {trackerId!r}")
if trackerId_int is None:
return ActionResult.isFailure(error="trackerId is required")
try:
payload = RedmineTicketCreateRequest(
subject=subject,
trackerId=trackerId_int,
description=parameters.get("description") or "",
statusId=_optInt(parameters.get("statusId")),
priorityId=_optInt(parameters.get("priorityId")),
assignedToId=_optInt(parameters.get("assignedToId")),
parentIssueId=_optInt(parameters.get("parentIssueId")),
fixedVersionId=_optInt(parameters.get("fixedVersionId")),
customFields=parameters.get("customFields") or None,
)
except Exception as exc:
return ActionResult.isFailure(error=f"Invalid create body: {exc}")
try:
created = await createTicket(user, mandateId, featureInstanceId, payload)
except Exception as exc:
logger.exception("redmine.createTicket failed")
return ActionResult.isFailure(error=f"Create ticket failed: {exc}")
return ActionResult.isSuccess(data={"ticket": ticketToDict(created)})
def _optInt(value: Any):
if value is None or value == "":
return None
try:
return int(value)
except (TypeError, ValueError):
return None