retoor/rp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: standardize string quotes and fix import ordering across multiple modules

Browse Source
This commit is contained in:
retoor
2025-11-04 07:09:12 +00:00
parent ea29bdc403
commit e9ced4a493
82 changed files with 4963 additions and 3094 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pr/workflows/__init__.py
+8 -2
View File
@@ -1,5 +1,11 @@
from .workflow_definition import Workflow, WorkflowStep, ExecutionMode
from .workflow_definition import ExecutionMode, Workflow, WorkflowStep
from .workflow_engine import WorkflowEngine
from .workflow_storage import WorkflowStorage
__all__ = ['Workflow', 'WorkflowStep', 'ExecutionMode', 'WorkflowEngine', 'WorkflowStorage']
__all__ = [
"Workflow",
"WorkflowStep",
"ExecutionMode",
"WorkflowEngine",
"WorkflowStorage",
]
pr/workflows/workflow_definition.py
+35 -32
View File
@@ -1,12 +1,14 @@
from enum import Enum
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Dict, List, Optional
class ExecutionMode(Enum):
SEQUENTIAL = "sequential"
PARALLEL = "parallel"
CONDITIONAL = "conditional"
@dataclass
class WorkflowStep:
tool_name: str
@@ -20,29 +22,30 @@ class WorkflowStep:
def to_dict(self) -> Dict[str, Any]:
return {
'tool_name': self.tool_name,
'arguments': self.arguments,
'step_id': self.step_id,
'condition': self.condition,
'on_success': self.on_success,
'on_failure': self.on_failure,
'retry_count': self.retry_count,
'timeout_seconds': self.timeout_seconds
"tool_name": self.tool_name,
"arguments": self.arguments,
"step_id": self.step_id,
"condition": self.condition,
"on_success": self.on_success,
"on_failure": self.on_failure,
"retry_count": self.retry_count,
"timeout_seconds": self.timeout_seconds,
}
@staticmethod
def from_dict(data: Dict[str, Any]) -> 'WorkflowStep':
def from_dict(data: Dict[str, Any]) -> "WorkflowStep":
return WorkflowStep(
tool_name=data['tool_name'],
arguments=data['arguments'],
step_id=data['step_id'],
condition=data.get('condition'),
on_success=data.get('on_success'),
on_failure=data.get('on_failure'),
retry_count=data.get('retry_count', 0),
timeout_seconds=data.get('timeout_seconds', 300)
tool_name=data["tool_name"],
arguments=data["arguments"],
step_id=data["step_id"],
condition=data.get("condition"),
on_success=data.get("on_success"),
on_failure=data.get("on_failure"),
retry_count=data.get("retry_count", 0),
timeout_seconds=data.get("timeout_seconds", 300),
)
@dataclass
class Workflow:
name: str
@@ -54,23 +57,23 @@ class Workflow:
def to_dict(self) -> Dict[str, Any]:
return {
'name': self.name,
'description': self.description,
'steps': [step.to_dict() for step in self.steps],
'execution_mode': self.execution_mode.value,
'variables': self.variables,
'tags': self.tags
"name": self.name,
"description": self.description,
"steps": [step.to_dict() for step in self.steps],
"execution_mode": self.execution_mode.value,
"variables": self.variables,
"tags": self.tags,
}
@staticmethod
def from_dict(data: Dict[str, Any]) -> 'Workflow':
def from_dict(data: Dict[str, Any]) -> "Workflow":
return Workflow(
name=data['name'],
description=data['description'],
steps=[WorkflowStep.from_dict(step) for step in data['steps']],
execution_mode=ExecutionMode(data.get('execution_mode', 'sequential')),
variables=data.get('variables', {}),
tags=data.get('tags', [])
name=data["name"],
description=data["description"],
steps=[WorkflowStep.from_dict(step) for step in data["steps"]],
execution_mode=ExecutionMode(data.get("execution_mode", "sequential")),
variables=data.get("variables", {}),
tags=data.get("tags", []),
)
def add_step(self, step: WorkflowStep):
pr/workflows/workflow_engine.py
+81 -53
View File
@@ -1,8 +1,10 @@
import time
import re
from typing import Dict, Any, List, Callable, Optional
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from .workflow_definition import Workflow, WorkflowStep, ExecutionMode
from typing import Any, Callable, Dict, List, Optional
from .workflow_definition import ExecutionMode, Workflow, WorkflowStep
class WorkflowExecutionContext:
def __init__(self):
@@ -23,57 +25,66 @@ class WorkflowExecutionContext:
return self.step_results.get(step_id)
def log_event(self, event_type: str, step_id: str, details: Dict[str, Any]):
self.execution_log.append({
'timestamp': time.time(),
'event_type': event_type,
'step_id': step_id,
'details': details
})
self.execution_log.append(
{
"timestamp": time.time(),
"event_type": event_type,
"step_id": step_id,
"details": details,
}
)
class WorkflowEngine:
def __init__(self, tool_executor: Callable, max_workers: int = 5):
self.tool_executor = tool_executor
self.max_workers = max_workers
def _evaluate_condition(self, condition: str, context: WorkflowExecutionContext) -> bool:
def _evaluate_condition(
self, condition: str, context: WorkflowExecutionContext
) -> bool:
if not condition:
return True
try:
safe_locals = {
'variables': context.variables,
'results': context.step_results
"variables": context.variables,
"results": context.step_results,
}
return eval(condition, {"__builtins__": {}}, safe_locals)
except Exception:
return False
def _substitute_variables(self, arguments: Dict[str, Any], context: WorkflowExecutionContext) -> Dict[str, Any]:
def _substitute_variables(
self, arguments: Dict[str, Any], context: WorkflowExecutionContext
) -> Dict[str, Any]:
substituted = {}
for key, value in arguments.items():
if isinstance(value, str):
pattern = r'\$\{([^}]+)\}'
pattern = r"\$\{([^}]+)\}"
matches = re.findall(pattern, value)
for match in matches:
if match.startswith('step.'):
step_id = match.split('.', 1)[1]
if match.startswith("step."):
step_id = match.split(".", 1)[1]
replacement = context.get_step_result(step_id)
if replacement is not None:
value = value.replace(f'${{{match}}}', str(replacement))
elif match.startswith('var.'):
var_name = match.split('.', 1)[1]
value = value.replace(f"${{{match}}}", str(replacement))
elif match.startswith("var."):
var_name = match.split(".", 1)[1]
replacement = context.get_variable(var_name)
if replacement is not None:
value = value.replace(f'${{{match}}}', str(replacement))
value = value.replace(f"${{{match}}}", str(replacement))
substituted[key] = value
else:
substituted[key] = value
return substituted
def _execute_step(self, step: WorkflowStep, context: WorkflowExecutionContext) -> Dict[str, Any]:
def _execute_step(
self, step: WorkflowStep, context: WorkflowExecutionContext
) -> Dict[str, Any]:
if not self._evaluate_condition(step.condition, context):
context.log_event('skipped', step.step_id, {'reason': 'condition_not_met'})
return {'status': 'skipped', 'step_id': step.step_id}
context.log_event("skipped", step.step_id, {"reason": "condition_not_met"})
return {"status": "skipped", "step_id": step.step_id}
arguments = self._substitute_variables(step.arguments, context)
@@ -83,26 +94,34 @@ class WorkflowEngine:
while retry_attempts <= step.retry_count:
try:
context.log_event('executing', step.step_id, {
'tool': step.tool_name,
'arguments': arguments,
'attempt': retry_attempts + 1
})
context.log_event(
"executing",
step.step_id,
{
"tool": step.tool_name,
"arguments": arguments,
"attempt": retry_attempts + 1,
},
)
result = self.tool_executor(step.tool_name, arguments)
execution_time = time.time() - start_time
context.set_step_result(step.step_id, result)
context.log_event('completed', step.step_id, {
'execution_time': execution_time,
'result_size': len(str(result)) if result else 0
})
context.log_event(
"completed",
step.step_id,
{
"execution_time": execution_time,
"result_size": len(str(result)) if result else 0,
},
)
return {
'status': 'success',
'step_id': step.step_id,
'result': result,
'execution_time': execution_time
"status": "success",
"step_id": step.step_id,
"result": result,
"execution_time": execution_time,
}
except Exception as e:
@@ -111,25 +130,26 @@ class WorkflowEngine:
if retry_attempts <= step.retry_count:
time.sleep(1 * retry_attempts)
context.log_event('failed', step.step_id, {'error': last_error})
context.log_event("failed", step.step_id, {"error": last_error})
return {
'status': 'failed',
'step_id': step.step_id,
'error': last_error,
'execution_time': time.time() - start_time
"status": "failed",
"step_id": step.step_id,
"error": last_error,
"execution_time": time.time() - start_time,
}
def _get_next_steps(self, completed_step: WorkflowStep, result: Dict[str, Any],
workflow: Workflow) -> List[WorkflowStep]:
def _get_next_steps(
self, completed_step: WorkflowStep, result: Dict[str, Any], workflow: Workflow
) -> List[WorkflowStep]:
next_steps = []
if result['status'] == 'success' and completed_step.on_success:
if result["status"] == "success" and completed_step.on_success:
for step_id in completed_step.on_success:
step = workflow.get_step(step_id)
if step:
next_steps.append(step)
elif result['status'] == 'failed' and completed_step.on_failure:
elif result["status"] == "failed" and completed_step.on_failure:
for step_id in completed_step.on_failure:
step = workflow.get_step(step_id)
if step:
@@ -142,7 +162,9 @@ class WorkflowEngine:
return next_steps
def execute_workflow(self, workflow: Workflow, initial_variables: Optional[Dict[str, Any]] = None) -> WorkflowExecutionContext:
def execute_workflow(
self, workflow: Workflow, initial_variables: Optional[Dict[str, Any]] = None
) -> WorkflowExecutionContext:
context = WorkflowExecutionContext()
if initial_variables:
@@ -151,7 +173,7 @@ class WorkflowEngine:
if workflow.variables:
context.variables.update(workflow.variables)
context.log_event('workflow_started', 'workflow', {'name': workflow.name})
context.log_event("workflow_started", "workflow", {"name": workflow.name})
if workflow.execution_mode == ExecutionMode.PARALLEL:
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
@@ -164,9 +186,11 @@ class WorkflowEngine:
step = futures[future]
try:
result = future.result()
context.log_event('step_completed', step.step_id, result)
context.log_event("step_completed", step.step_id, result)
except Exception as e:
context.log_event('step_failed', step.step_id, {'error': str(e)})
context.log_event(
"step_failed", step.step_id, {"error": str(e)}
)
else:
pending_steps = workflow.get_initial_steps()
@@ -184,9 +208,13 @@ class WorkflowEngine:
next_steps = self._get_next_steps(step, result, workflow)
pending_steps.extend(next_steps)
context.log_event('workflow_completed', 'workflow', {
'total_steps': len(context.step_results),
'executed_steps': list(context.step_results.keys())
})
context.log_event(
"workflow_completed",
"workflow",
{
"total_steps": len(context.step_results),
"executed_steps": list(context.step_results.keys()),
},
)
return context
pr/workflows/workflow_storage.py
+102 -53
View File
@@ -2,8 +2,10 @@ import json
import sqlite3
import time
from typing import List, Optional
from .workflow_definition import Workflow
class WorkflowStorage:
def __init__(self, db_path: str):
self.db_path = db_path
@@ -13,7 +15,8 @@ class WorkflowStorage:
conn = sqlite3.connect(self.db_path, check_same_thread=False)
cursor = conn.cursor()
cursor.execute('''
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS workflows (
workflow_id TEXT PRIMARY KEY,
name TEXT NOT NULL,
@@ -25,9 +28,11 @@ class WorkflowStorage:
last_execution_at INTEGER,
tags TEXT
)
''')
"""
)
cursor.execute('''
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS workflow_executions (
execution_id TEXT PRIMARY KEY,
workflow_id TEXT NOT NULL,
@@ -39,17 +44,24 @@ class WorkflowStorage:
step_results TEXT,
FOREIGN KEY (workflow_id) REFERENCES workflows(workflow_id)
)
''')
"""
)
cursor.execute('''
cursor.execute(
"""
CREATE INDEX IF NOT EXISTS idx_workflow_name ON workflows(name)
''')
cursor.execute('''
"""
)
cursor.execute(
"""
CREATE INDEX IF NOT EXISTS idx_execution_workflow ON workflow_executions(workflow_id)
''')
cursor.execute('''
"""
)
cursor.execute(
"""
CREATE INDEX IF NOT EXISTS idx_execution_started ON workflow_executions(started_at)
''')
"""
)
conn.commit()
conn.close()
@@ -66,12 +78,22 @@ class WorkflowStorage:
current_time = int(time.time())
tags_json = json.dumps(workflow.tags)
cursor.execute('''
cursor.execute(
"""
INSERT OR REPLACE INTO workflows
(workflow_id, name, description, workflow_data, created_at, updated_at, tags)
VALUES (?, ?, ?, ?, ?, ?, ?)
''', (workflow_id, workflow.name, workflow.description, workflow_data,
current_time, current_time, tags_json))
""",
(
workflow_id,
workflow.name,
workflow.description,
workflow_data,
current_time,
current_time,
tags_json,
),
)
conn.commit()
conn.close()
@@ -82,7 +104,9 @@ class WorkflowStorage:
conn = sqlite3.connect(self.db_path, check_same_thread=False)
cursor = conn.cursor()
cursor.execute('SELECT workflow_data FROM workflows WHERE workflow_id = ?', (workflow_id,))
cursor.execute(
"SELECT workflow_data FROM workflows WHERE workflow_id = ?", (workflow_id,)
)
row = cursor.fetchone()
conn.close()
@@ -95,7 +119,7 @@ class WorkflowStorage:
conn = sqlite3.connect(self.db_path, check_same_thread=False)
cursor = conn.cursor()
cursor.execute('SELECT workflow_data FROM workflows WHERE name = ?', (name,))
cursor.execute("SELECT workflow_data FROM workflows WHERE name = ?", (name,))
row = cursor.fetchone()
conn.close()
@@ -109,29 +133,36 @@ class WorkflowStorage:
cursor = conn.cursor()
if tag:
cursor.execute('''
cursor.execute(
"""
SELECT workflow_id, name, description, execution_count, last_execution_at, tags
FROM workflows
WHERE tags LIKE ?
ORDER BY name
''', (f'%"{tag}"%',))
""",
(f'%"{tag}"%',),
)
else:
cursor.execute('''
cursor.execute(
"""
SELECT workflow_id, name, description, execution_count, last_execution_at, tags
FROM workflows
ORDER BY name
''')
"""
)
workflows = []
for row in cursor.fetchall():
workflows.append({
'workflow_id': row[0],
'name': row[1],
'description': row[2],
'execution_count': row[3],
'last_execution_at': row[4],
'tags': json.loads(row[5]) if row[5] else []
})
workflows.append(
{
"workflow_id": row[0],
"name": row[1],
"description": row[2],
"execution_count": row[3],
"last_execution_at": row[4],
"tags": json.loads(row[5]) if row[5] else [],
}
)
conn.close()
return workflows
@@ -140,18 +171,21 @@ class WorkflowStorage:
conn = sqlite3.connect(self.db_path, check_same_thread=False)
cursor = conn.cursor()
cursor.execute('DELETE FROM workflows WHERE workflow_id = ?', (workflow_id,))
cursor.execute("DELETE FROM workflows WHERE workflow_id = ?", (workflow_id,))
deleted = cursor.rowcount > 0
cursor.execute('DELETE FROM workflow_executions WHERE workflow_id = ?', (workflow_id,))
cursor.execute(
"DELETE FROM workflow_executions WHERE workflow_id = ?", (workflow_id,)
)
conn.commit()
conn.close()
return deleted
def save_execution(self, workflow_id: str, execution_context: 'WorkflowExecutionContext') -> str:
import hashlib
def save_execution(
self, workflow_id: str, execution_context: "WorkflowExecutionContext"
) -> str:
import uuid
execution_id = str(uuid.uuid4())[:16]
@@ -159,30 +193,40 @@ class WorkflowStorage:
conn = sqlite3.connect(self.db_path, check_same_thread=False)
cursor = conn.cursor()
started_at = int(execution_context.execution_log[0]['timestamp']) if execution_context.execution_log else int(time.time())
started_at = (
int(execution_context.execution_log[0]["timestamp"])
if execution_context.execution_log
else int(time.time())
)
completed_at = int(time.time())
cursor.execute('''
cursor.execute(
"""
INSERT INTO workflow_executions
(execution_id, workflow_id, started_at, completed_at, status, execution_log, variables, step_results)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
''', (
execution_id,
workflow_id,
started_at,
completed_at,
'completed',
json.dumps(execution_context.execution_log),
json.dumps(execution_context.variables),
json.dumps(execution_context.step_results)
))
""",
(
execution_id,
workflow_id,
started_at,
completed_at,
"completed",
json.dumps(execution_context.execution_log),
json.dumps(execution_context.variables),
json.dumps(execution_context.step_results),
),
)
cursor.execute('''
cursor.execute(
"""
UPDATE workflows
SET execution_count = execution_count + 1,
last_execution_at = ?
WHERE workflow_id = ?
''', (completed_at, workflow_id))
""",
(completed_at, workflow_id),
)
conn.commit()
conn.close()
@@ -193,22 +237,27 @@ class WorkflowStorage:
conn = sqlite3.connect(self.db_path, check_same_thread=False)
cursor = conn.cursor()
cursor.execute('''
cursor.execute(
"""
SELECT execution_id, started_at, completed_at, status
FROM workflow_executions
WHERE workflow_id = ?
ORDER BY started_at DESC
LIMIT ?
''', (workflow_id, limit))
""",
(workflow_id, limit),
)
executions = []
for row in cursor.fetchall():
executions.append({
'execution_id': row[0],
'started_at': row[1],
'completed_at': row[2],
'status': row[3]
})
executions.append(
{
"execution_id": row[0],
"started_at": row[1],
"completed_at": row[2],
"status": row[3],
}
)
conn.close()
return executions