workflow.py 1.7 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465
  1. from typing import List, Optional
  2. from pydantic import BaseModel, ConfigDict, field_validator
  3. from pydantic.alias_generators import to_camel
  4. def _coerce_optional_id(value):
  5. if value is None:
  6. return None
  7. return str(value).strip()
  8. class WorkflowNodeCreate(BaseModel):
  9. model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)
  10. node_name: str
  11. node_order: int
  12. approver_id: Optional[str] = None
  13. approver_name: Optional[str] = None
  14. action_type: Optional[str] = "approve"
  15. @field_validator("approver_id", mode="before")
  16. @classmethod
  17. def validate_approver_id(cls, value):
  18. return _coerce_optional_id(value)
  19. class WorkflowNodeOut(BaseModel):
  20. model_config = ConfigDict(from_attributes=True, populate_by_name=True, alias_generator=to_camel)
  21. id: str
  22. node_name: str
  23. node_order: int
  24. approver_id: Optional[str] = None
  25. approver_name: Optional[str] = None
  26. action_type: Optional[str] = "approve"
  27. class WorkflowDefCreate(BaseModel):
  28. model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)
  29. biz_type: str
  30. wf_name: str
  31. status: int = 1
  32. nodes: Optional[List[WorkflowNodeCreate]] = None
  33. class WorkflowDefUpdate(BaseModel):
  34. model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)
  35. biz_type: Optional[str] = None
  36. wf_name: Optional[str] = None
  37. status: Optional[int] = None
  38. nodes: Optional[List[WorkflowNodeCreate]] = None
  39. class WorkflowDefOut(BaseModel):
  40. model_config = ConfigDict(from_attributes=True, populate_by_name=True, alias_generator=to_camel)
  41. id: str
  42. biz_type: str
  43. wf_name: str
  44. status: int
  45. create_by: Optional[str] = None
  46. nodes: List[WorkflowNodeOut] = []