mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-11-09 05:24:49 +09:00
Juan Denis
d2d87ecc1d
Introduce a full-featured workflow automation system: add WorkflowEngine service to execute workflows, background scheduler to trigger cron workflows, and croniter dependency. Extend Workflow model with automation fields (is_active, trigger_type, trigger_config, last_run_at, last_status) and add WorkflowExecution and WorkflowLog models. Add API endpoints to start executions, list executions, fetch execution details and logs, and persist execution records/logs. Implement frontend components for workflow nodes, config panels, and an execution history UI (multiple new React components) and wire minor builder UI imports. Note: database migrations are required to add new columns and tables.
108 lines
4.6 KiB
Python
108 lines
4.6 KiB
Python
from datetime import datetime
|
|
from app import db
|
|
import json
|
|
|
|
|
|
class Workflow(db.Model):
|
|
__tablename__ = 'workflows'
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
name = db.Column(db.String(100), nullable=False)
|
|
description = db.Column(db.Text, nullable=True)
|
|
|
|
# React Flow state stored as JSON strings
|
|
nodes = db.Column(db.Text, nullable=True) # JSON array of nodes
|
|
edges = db.Column(db.Text, nullable=True) # JSON array of edges
|
|
viewport = db.Column(db.Text, nullable=True) # JSON object {x, y, zoom}
|
|
|
|
# Automation fields
|
|
is_active = db.Column(db.Boolean, default=False)
|
|
trigger_type = db.Column(db.String(50), default='manual') # manual, cron, event, webhook
|
|
trigger_config = db.Column(db.Text, nullable=True) # JSON configuration for the trigger
|
|
last_run_at = db.Column(db.DateTime, nullable=True)
|
|
last_status = db.Column(db.String(20), nullable=True) # success, failed
|
|
|
|
# Metadata
|
|
created_at = db.Column(db.DateTime, default=datetime.utcnow)
|
|
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
|
|
|
# Foreign keys
|
|
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
|
|
|
|
# Relationships
|
|
user = db.relationship('User', backref=db.backref('workflows', lazy='dynamic'))
|
|
executions = db.relationship('WorkflowExecution', backref='workflow', lazy='dynamic', cascade='all, delete-orphan')
|
|
|
|
def to_dict(self):
|
|
return {
|
|
'id': self.id,
|
|
'name': self.name,
|
|
'description': self.description,
|
|
'nodes': json.loads(self.nodes) if self.nodes else [],
|
|
'edges': json.loads(self.edges) if self.edges else [],
|
|
'viewport': json.loads(self.viewport) if self.viewport else None,
|
|
'is_active': self.is_active,
|
|
'trigger_type': self.trigger_type,
|
|
'trigger_config': json.loads(self.trigger_config) if self.trigger_config else {},
|
|
'last_run_at': self.last_run_at.isoformat() if self.last_run_at else None,
|
|
'last_status': self.last_status,
|
|
'created_at': self.created_at.isoformat() if self.created_at else None,
|
|
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
|
|
'user_id': self.user_id,
|
|
'node_count': len(json.loads(self.nodes)) if self.nodes else 0,
|
|
'edge_count': len(json.loads(self.edges)) if self.edges else 0
|
|
}
|
|
|
|
def __repr__(self):
|
|
return f'<Workflow {self.name}>'
|
|
|
|
|
|
class WorkflowExecution(db.Model):
|
|
__tablename__ = 'workflow_executions'
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
workflow_id = db.Column(db.Integer, db.ForeignKey('workflows.id'), nullable=False)
|
|
status = db.Column(db.String(20), default='running') # running, success, failed, cancelled
|
|
trigger_type = db.Column(db.String(50))
|
|
context = db.Column(db.Text, nullable=True) # JSON data passed between steps
|
|
results = db.Column(db.Text, nullable=True) # JSON results of each step
|
|
|
|
started_at = db.Column(db.DateTime, default=datetime.utcnow)
|
|
completed_at = db.Column(db.DateTime, nullable=True)
|
|
|
|
logs = db.relationship('WorkflowLog', backref='execution', lazy='dynamic', cascade='all, delete-orphan')
|
|
|
|
def to_dict(self):
|
|
return {
|
|
'id': self.id,
|
|
'workflow_id': self.workflow_id,
|
|
'status': self.status,
|
|
'trigger_type': self.trigger_type,
|
|
'context': json.loads(self.context) if self.context else {},
|
|
'results': json.loads(self.results) if self.results else {},
|
|
'started_at': self.started_at.isoformat() if self.started_at else None,
|
|
'completed_at': self.completed_at.isoformat() if self.completed_at else None,
|
|
'duration': (self.completed_at - self.started_at).total_seconds() if self.completed_at else None
|
|
}
|
|
|
|
|
|
class WorkflowLog(db.Model):
|
|
__tablename__ = 'workflow_logs'
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
execution_id = db.Column(db.Integer, db.ForeignKey('workflow_executions.id'), nullable=False)
|
|
level = db.Column(db.String(10), default='INFO') # INFO, WARNING, ERROR, DEBUG
|
|
message = db.Column(db.Text, nullable=False)
|
|
node_id = db.Column(db.String(100), nullable=True) # ID of the node that generated the log
|
|
timestamp = db.Column(db.DateTime, default=datetime.utcnow)
|
|
|
|
def to_dict(self):
|
|
return {
|
|
'id': self.id,
|
|
'execution_id': self.execution_id,
|
|
'level': self.level,
|
|
'message': self.message,
|
|
'node_id': self.node_id,
|
|
'timestamp': self.timestamp.isoformat()
|
|
}
|