aiskillstore HappyFlow Generator Skill
Metadata
- Skill Name: HappyFlow Generator
- Version: 2.0.0
- Category: API Testing & Automation
- Required Capabilities: Code execution, web requests, file operations
- Estimated Duration: 2-5 minutes per API spec
- Difficulty: Intermediate
Description
Automatically generate and execute Python test scripts from OpenAPI specifications and GraphQL schemas that successfully call all API endpoints in dependency-correct order, ensuring all requests return 2xx status codes.
Input: OpenAPI/GraphQL spec (URL/file) + authentication credentials
Output: Working Python script that executes complete API happy path flow
Key Features:
- Multi-format support: OpenAPI 3.0+ and GraphQL schemas
- Enhanced execution: Parallel execution, detailed reporting, connection pooling
- Advanced testing: File upload support, response schema validation, rate limiting handling
- Modular architecture: Well-organized codebase with proper error handling
Complete Workflow
Phase 1: Authentication Setup
Execute this code to prepare authentication headers:
import base64
import requests
from typing import Dict, Any
def setup_authentication(auth_type: str, credentials: Dict[str, Any]) -> Dict[str, str]:
"""Prepare authentication headers based on auth type"""
if auth_type == "bearer":
return {"Authorization": f"Bearer {credentials['token']}"}
elif auth_type == "api_key":
header_name = credentials.get('header_name', 'X-API-Key')
return {header_name: credentials['api_key']}
elif auth_type == "basic":
auth_string = f"{credentials['username']}:{credentials['password']}"
encoded = base64.b64encode(auth_string.encode()).decode()
return {"Authorization": f"Basic {encoded}"}
elif auth_type == "oauth2_client_credentials":
token_url = credentials['token_url']
data = {
'grant_type': 'client_credentials',
'client_id': credentials['client_id'],
'client_secret': credentials['client_secret']
}
if 'scopes' in credentials:
data['scope'] = ' '.join(credentials['scopes'])
response = requests.post(token_url, data=data)
response.raise_for_status()
token_data = response.json()
return {"Authorization": f"Bearer {token_data['access_token']}"}
return {}
# Example usage:
# auth_headers = setup_authentication("bearer", {"token": "abc123"})
Phase 2: Specification Parsing
Execute this code to parse API specifications (OpenAPI or GraphQL):
import requests
import yaml
import json
import re
from typing import Dict, List, Any, Union
from pathlib import Path
def parse_specification(spec_source: Union[str, Path], spec_type: str = "auto", **kwargs) -> Dict[str, Any]:
"""Parse API specification and extract structured information
Args:
spec_source: Path or URL to API specification
spec_type: Type of specification ('openapi', 'graphql', or 'auto')
**kwargs: Additional arguments for specific parsers
Returns:
Dictionary containing parsed specification data
"""
# Auto-detect specification type if not specified
if spec_type == "auto":
if isinstance(spec_source, str):
if spec_source.endswith(".graphql") or "graphql" in spec_source.lower():
spec_type = "graphql"
else:
spec_type = "openapi"
else:
# For file paths, check extension
path = Path(spec_source)
if path.suffix.lower() in [".graphql", ".gql"]:
spec_type = "graphql"
else:
spec_type = "openapi"
# Parse based on detected type
if spec_type == "openapi":
return parse_openapi_spec(spec_source, **kwargs)
elif spec_type == "graphql":
return parse_graphql_spec(spec_source, **kwargs)
else:
raise ValueError(f"Unsupported specification type: {spec_type}")
def parse_openapi_spec(spec_source: Union[str, Path], headers: Dict[str, str] = None) -> Dict[str, Any]:
"""Parse OpenAPI specification and extract structured information"""
# Fetch spec
if isinstance(spec_source, str) and spec_source.startswith('http'):
response = requests.get(spec_source, headers=headers or {})
response.raise_for_status()
content = response.text
try:
spec = json.loads(content)
except json.JSONDecodeError:
spec = yaml.safe_load(content)
else:
with open(spec_source, 'r') as f:
content = f.read()
try:
spec = json.loads(content)
except json.JSONDecodeError:
spec = yaml.safe_load(content)
# Extract base information
openapi_version = spec.get('openapi', spec.get('swagger', 'unknown'))
base_url = ""
if 'servers' in spec and spec['servers']:
base_url = spec['servers'][0]['url']
elif 'host' in spec:
scheme = spec.get('schemes', ['https'])[0]
base_path = spec.get('basePath', '')
base_url = f"{scheme}://{spec['host']}{base_path}"
# Extract endpoints
endpoints = []
paths = spec.get('paths', {})
for path, path_item in paths.items():
for method in ['get', 'post', 'put', 'patch', 'delete']:
if method not in path_item:
continue
operation = path_item[method]
# Extract parameters
parameters = []
for param in operation.get('parameters', []):
parameters.append({
'name': param.get('name'),
'in': param.get('in'),
'required': param.get('required', False),
'schema': param.get('schema', {}),
'example': param.get('example')
})
# Extract request body
request_body = None
if 'requestBody' in operation:
rb = operation['requestBody']
content = rb.get('content', {})
if 'application/json' in content:
json_content = content['application/json']
request_body = {
'required': rb.get('required', False),
'content_type': 'application/json',
'schema': json_content.get('schema', {}),
'example': json_content.get('example')
}
elif 'multipart/form-data' in content:
form_content = content['multipart/form-data']
request_body = {
'required': rb.get('required', False),
'content_type': 'multipart/form-data',
'schema': form_content.get('schema', {}),
'example': form_content.get('example')
}
# Extract responses
responses = {}
for status_code, response_data in operation.get('responses', {}).items():
if status_code.startswith('2'):
content = response_data.get('content', {})
if 'application/json' in content:
json_content = content['application/json']
responses[status_code] = {
'description': response_data.get('description', ''),
'schema': json_content.get('schema', {}),
'example': json_content.get('example')
}
endpoint = {
'operation_id': operation.get('operationId', f"{method}_{path}"),
'path': path,
'method': method.upper(),
'tags': operation.get('tags', []),
'summary': operation.get('summary', ''),
'parameters': parameters,
'request_body': request_body,
'responses': responses
}
endpoints.append(endpoint)
return {
'openapi_version': openapi_version,
'base_url': base_url,
'endpoints': endpoints,
'schemas': spec.get('components', {}).get('schemas', {})
}
def parse_graphql_spec(spec_source: str, headers: Dict[str, str] = None) -> Dict[str, Any]:
"""Parse GraphQL schema and extract operations"""
# For GraphQL, we'll create a simplified representation
# In practice, this would use graphql-core to parse the schema
base_url = spec_source if isinstance(spec_source, str) and spec_source.startswith('http') else ""
# Placeholder for GraphQL endpoints - in reality, this would be derived from schema introspection
endpoints = [
{
'operation_id': 'graphql_query',
'path': '/graphql',
'method': 'POST',
'tags': ['GraphQL'],
'summary': 'GraphQL Query',
'parameters': [],
'request_body': {
'required': True,
'content_type': 'application/json',
'schema': {},
'example': {'query': 'query { __schema { types { name } } }'}
},
'responses': {
'200': {
'description': 'Successful GraphQL response',
'schema': {},
'example': {}
}
}
}
]
return {
'spec_type': 'graphql',
'base_url': base_url,
'endpoints': endpoints,
'schemas': {}
}
# Example usage:
# parsed_spec = parse_specification("https://api.example.com/openapi.json")
# parsed_spec = parse_specification("https://api.example.com/graphql", spec_type="graphql")
Phase 3: Dependency Analysis
Execute this code to analyze dependencies and determine execution order:
import re
from typing import List, Dict, Any
def analyze_dependencies(endpoints: List[Dict]) -> Dict[str, Any]:
"""Analyze endpoint dependencies and create execution order"""
dependencies = {}
outputs = {}
for endpoint in endpoints:
endpoint_id = f"{endpoint['method']} {endpoint['path']}"
dependencies[endpoint_id] = []
outputs[endpoint_id] = {}
# Detect path parameter dependencies
for endpoint in endpoints:
endpoint_id = f"{endpoint['method']} {endpoint['path']}"
path = endpoint['path']
path_params = re.findall(r'\{(\w+)\}', path)
for param in path_params:
for other_endpoint in endpoints:
other_id = f"{other_endpoint['method']} {other_endpoint['path']}"
if other_endpoint['method'] in ['POST', 'PUT']:
for status, response in other_endpoint.get('responses', {}).items():
schema = response.get('schema', {})
properties = schema.get('properties', {})
if 'id' in properties or param in properties:
if other_id != endpoint_id and other_id not in dependencies[endpoint_id]:
dependencies[endpoint_id].append(other_id)
output_field = 'id' if 'id' in properties else param
outputs[other_id][param] = f"response.body.{output_field}"
# HTTP method ordering
method_priority = {'POST': 1, 'GET': 2, 'PUT': 3, 'PATCH': 3, 'DELETE': 4}
for endpoint in endpoints:
endpoint_id = f"{endpoint['method']} {endpoint['path']}"
path_clean = re.sub(r'\{[^}]+\}', '', endpoint['path'])
for other_endpoint in endpoints:
other_id = f"{other_endpoint['method']} {other_endpoint['path']}"
other_path_clean = re.sub(r'\{[^}]+\}', '', other_endpoint['path'])
if path_clean == other_path_clean:
if method_priority.get(endpoint['method'], 5) > method_priority.get(other_endpoint['method'], 5):
if other_id not in dependencies[endpoint_id]:
dependencies[endpoint_id].append(other_id)
# Topological sort
def topological_sort(deps):
in_degree = {node: 0 for node in deps}
for node in deps:
for dep in deps[node]:
in_degree[dep] = in_degree.get(dep, 0) + 1
queue = [node for node in deps if in_degree[node] == 0]
result = []
while queue:
queue.sort(key=lambda x: (x.split()[1].count('/'), method_priority.get(x.split()[0], 5)))
node = queue.pop(0)
result.append(node)
for other_node in deps:
if node in deps[other_node]:
in_degree[other_node] -= 1
if in_degree[other_node] == 0:
queue.append(other_node)
return result
execution_order_ids = topological_sort(dependencies)
execution_plan = []
for step, endpoint_id in enumerate(execution_order_ids, 1):
endpoint = next(e for e in endpoints if f"{e['method']} {e['path']}" == endpoint_id)
inputs = {}
for dep_id in dependencies[endpoint_id]:
if dep_id in outputs:
for param_name, json_path in outputs[dep_id].items():
dep_step = execution_order_ids.index(dep_id) + 1
inputs[param_name] = {
'source': f"step_{dep_step}",
'json_path': json_path
}
execution_plan.append({
'step': step,
'endpoint': endpoint,
'dependencies': dependencies[endpoint_id],
'inputs': inputs,
'outputs': outputs[endpoint_id]
})
return {
'execution_order': execution_plan,
'dependency_graph': dependencies
}
def identify_parallel_groups(execution_plan: List[Dict]) -> List[List[int]]:
"""Identify groups of steps that can be executed in parallel"""
# Group steps by their dependencies
parallel_groups = []
processed_steps = set()
# Find steps with no dependencies (can run in parallel)
independent_steps = [step['step'] for step in execution_plan if not step['dependencies']]
if independent_steps:
parallel_groups.append(independent_steps)
processed_steps.update(independent_steps)
# For remaining steps, group those with the same dependencies
remaining_steps = [step for step in execution_plan if step['step'] not in processed_steps]
# Simple grouping by dependency sets
dependency_map = {}
for step in remaining_steps:
dep_tuple = tuple(sorted(step['dependencies']))
if dep_tuple not in dependency_map:
dependency_map[dep_tuple] = []
dependency_map[dep_tuple].append(step['step'])
for group in dependency_map.values():
parallel_groups.append(group)
return parallel_groups
# Example usage:
# dependency_analysis = analyze_dependencies(parsed_spec['endpoints'])
# parallel_groups = identify_parallel_groups(dependency_analysis['execution_order'])
Phase 4: Script Generation
Execute this code to generate the Python test script:
import json
import time
from typing import Dict, List, Any
from jsonschema import validate, ValidationError
def generate_value_from_schema(schema: Dict, field_name: str = "") -> Any:
"""Generate example value based on schema"""
if 'example' in schema:
return schema['example']
if 'default' in schema:
return schema['default']
if 'enum' in schema:
return schema['enum'][0]
schema_type = schema.get('type', 'string')
if schema_type == 'string':
if schema.get('format') == 'email':
return 'test@example.com'
elif schema.get('format') == 'uuid':
return '550e8400-e29b-41d4-a716-446655440000'
elif 'email' in field_name.lower():
return 'test@example.com'
elif 'name' in field_name.lower():
return 'Test User'
elif 'description' in field_name.lower():
return 'Test description'
return 'test_value'
elif schema_type == 'integer':
minimum = schema.get('minimum', 1)
maximum = schema.get('maximum', minimum + 100)
return max(minimum, 1) # Ensure positive for IDs
elif schema_type == 'number':
return 10.5
elif schema_type == 'boolean':
return True
elif schema_type == 'array':
items_schema = schema.get('items', {})
return [generate_value_from_schema(items_schema)]
elif schema_type == 'object':
obj = {}
for prop, prop_schema in schema.get('properties', {}).items():
if prop in schema.get('required', []) or not schema.get('required'):
obj[prop] = generate_value_from_schema(prop_schema, prop)
return obj
return None
def generate_python_script(
execution_plan: List[Dict],
base_url: str,
auth_headers: Dict,
parallel_execution: bool = False,
parallel_groups: List[List[int]] = None
) -> str:
"""Generate complete Python script"""
lines = []
# Header
lines.append('#!/usr/bin/env python3')
lines.append('"""HappyFlow Generator - Auto-generated API test script"""')
lines.append('')
lines.append('import requests')
lines.append('import json')
lines.append('import sys')
lines.append('import time')
lines.append('from datetime import datetime')
if parallel_execution:
lines.append('from concurrent.futures import ThreadPoolExecutor, as_completed')
lines.append('from jsonschema import validate, ValidationError')
lines.append('')
# Class
lines.append('class APIFlowExecutor:')
lines.append(' def __init__(self, base_url, auth_headers):')
lines.append(' self.base_url = base_url.rstrip("/")')
lines.append(' self.session = requests.Session()')
lines.append(' self.session.headers.update(auth_headers)')
lines.append(' self.context = {}')
lines.append(' self.results = []')
lines.append('')
lines.append(' def log(self, message, level="INFO"):')
lines.append(' print(f"[{datetime.utcnow().isoformat()}] [{level}] {message}")')
lines.append('')
lines.append(' def _make_request(self, method, url, **kwargs):')
lines.append(' """Make HTTP request with retry logic for rate limiting"""')
lines.append(' max_retries = 3')
lines.append(' for attempt in range(max_retries):')
lines.append(' try:')
lines.append(' response = self.session.request(method, url, **kwargs)')
lines.append(' # Handle rate limiting')
lines.append(' if response.status_code == 429:')
lines.append(' if attempt < max_retries - 1:')
lines.append(' delay = 2 ** attempt # Exponential backoff')
lines.append(' self.log(f"Rate limited. Waiting {delay}s before retry...", "WARN")')
lines.append(' time.sleep(delay)')
lines.append(' continue')
lines.append(' return response')
lines.append(' except Exception as e:')
lines.append(' if attempt < max_retries - 1:')
lines.append(' delay = 2 ** attempt')
lines.append(' self.log(f"Request failed: {e}. Retrying in {delay}s...", "WARN")')
lines.append(' time.sleep(delay)')
lines.append(' else:')
lines.append(' raise')
lines.append('')
if parallel_execution and parallel_groups:
lines.append(' def execute_parallel_group(self, step_numbers):')
lines.append(' """Execute a group of steps in parallel"""')
lines.append(' with ThreadPoolExecutor(max_workers=5) as executor:')
lines.append(' future_to_step = {')
for group in parallel_groups:
if len(group) > 1: # Only create parallel execution for groups with multiple steps
for step_num in group:
lines.append(f' executor.submit(self.step_{step_num}): {step_num},')
break
lines.append(' }')
lines.append(' ')
lines.append(' for future in as_completed(future_to_step):')
lines.append(' step_num = future_to_step[future]')
lines.append(' try:')
lines.append(' future.result()')
lines.append(' self.log(f"Step {step_num} completed successfully")')
lines.append(' except Exception as e:')
lines.append(' self.log(f"Step {step_num} failed: {e}", "ERROR")')
lines.append(' raise')
lines.append('')
lines.append(' def execute_flow(self):')
lines.append(' try:')
# If parallel execution is enabled, organize steps by groups
if parallel_execution and parallel_groups:
executed_steps = set()
for i, group in enumerate(parallel_groups):
if len(group) > 1:
# Parallel group
lines.append(f' # Parallel Group {i+1}')
lines.append(f' self.log("Executing parallel group: {group}")')
lines.append(f' self.execute_parallel_group({group})')
executed_steps.update(group)
else:
# Sequential step
step_num = group[0]
if step_num not in executed_steps:
lines.append(f' self.step_{step_num}()')
executed_steps.add(step_num)
# Execute any remaining steps not covered by groups
for step_info in execution_plan:
step_num = step_info['step']
if step_num not in executed_steps:
lines.append(f' self.step_{step_num}()')
else:
# Sequential execution
for step_info in execution_plan:
lines.append(f' self.step_{step_info["step"]}()')
lines.append(' self.log("✓ All requests completed", "SUCCESS")')
lines.append(' return True')
lines.append(' except Exception as e:')
lines.append(' self.log(f"✗ Failed: {e}", "ERROR")')
lines.append(' return False')
lines.append('')
# Generate steps
for step_info in execution_plan:
endpoint = step_info['endpoint']
step_num = step_info['step']
method = endpoint['method']
path = endpoint['path']
lines.append(f' def step_{step_num}(self):')
lines.append(f' """Step {step_num}: {method} {path}"""')
lines.append(f' self.log("Step {step_num}: {method} {path}")')
# Initialize tracking variables
lines.append(' # Initialize tracking variables')
lines.append(' start_time = time.time()')
lines.append(' request_details = {')
lines.append(' "method": "%s",' % method)
lines.append(' "url": None,')
lines.append(' "headers": dict(self.session.headers),')
lines.append(' "payload": None')
lines.append(' }')
lines.append(' response_details = {')
lines.append(' "status_code": None,')
lines.append(' "headers": None,')
lines.append(' "body": None,')
lines.append(' "elapsed": None')
lines.append(' }')
lines.append(' error_details = None')
lines.append('')
lines.append(' try:')
# Build URL
url_expr = f'f"{{self.base_url}}{path}"'
# Replace path parameters
if '{' in path:
for param in re.findall(r'\{(\w+)\}', path):
url_expr = url_expr.replace(f'{{{param}}}', f'{{self.context.get("{param}", "UNKNOWN_{param}")}}')
lines.append(f' # Build URL with path parameters')
lines.append(f' url = {url_expr}')
lines.append(' request_details["url"] = url')
lines.append('')
# Handle request body
if endpoint.get('request_body'):
schema = endpoint['request_body'].get('schema', {})
example = endpoint['request_body'].get('example')
content_type = endpoint['request_body'].get('content_type', 'application/json')
if example:
payload = example
else:
payload = generate_value_from_schema(schema)
lines.append(f' # Handle request body ({content_type})')
if content_type == 'multipart/form-data':
lines.append(' # Handle file uploads')
lines.append(' files = {}')
lines.append(f' payload = {json.dumps(payload) if payload else {}}')
lines.append(' request_details["payload"] = payload')
lines.append(' response = self._make_request("%s", url, data=payload, files=files)' % method.lower())
else:
lines.append(f' payload = {json.dumps(payload) if payload else {}}')
lines.append(' request_details["payload"] = payload')
lines.append(' response = self._make_request("%s", url, json=payload)' % method.lower())
else:
lines.append(' # No request body')
lines.append(' response = self._make_request("%s", url)' % method.lower())
lines.append(' self.log(f"Status: {response.status_code}")')
lines.append(' if response.status_code not in [200, 201, 202, 204]:')
lines.append(' raise Exception(f"Unexpected status code: {response.status_code}")')
# Process response
lines.append(' if response.text:')
lines.append(' try:')
lines.append(' data = response.json()')
# Add response validation if schema exists
success_response = None
for status_code, resp_data in endpoint.get('responses', {}).items():
if status_code.startswith('2'):
success_response = resp_data
break
if success_response and success_response.get('schema'):
schema = success_response['schema']
lines.append(' # Validate response against schema')
lines.append(' schema = %s' % json.dumps(schema))
lines.append(' try:')
lines.append(' validate(instance=data, schema=schema)')
lines.append(' self.log("Response validated successfully against schema")')
lines.append(' except ValidationError as e:')
lines.append(' self.log(f"Response validation failed: {e.message}", "ERROR")')
lines.append(' self.log(f"Validation path: {' -> '.join(str(x) for x in e.absolute_path)}", "ERROR")')
# Extract outputs
if step_info['outputs']:
for output_name, json_path in step_info['outputs'].items():
field = json_path.split('.')[-1]
lines.append(f' self.context["{output_name}"] = data.get("{field}")')
lines.append(' except ValueError:')
lines.append(' self.log("Warning: Response is not valid JSON", "WARN")')
# Calculate execution time
lines.append('')
lines.append(' # Calculate execution time')
lines.append(' end_time = time.time()')
lines.append(' elapsed_time = end_time - start_time')
lines.append('')
# Capture response details
lines.append(' # Capture response details')
lines.append(' response_details.update({')
lines.append(' "status_code": response.status_code,')
lines.append(' "headers": dict(response.headers),')
lines.append(' "body": response.text[:1000] if response.text else "",')
lines.append(' "elapsed": elapsed_time')
lines.append(' })')
lines.append('')
lines.append(' except Exception as e:')
lines.append(' error_details = str(e)')
lines.append(' self.log(f"Error processing response: {e}", "ERROR")')
lines.append(' # Still capture timing info even on error')
lines.append(' end_time = time.time()')
lines.append(' elapsed_time = end_time - start_time if "start_time" in locals() else 0')
lines.append(' # Capture partial response details if available')
lines.append(' if "response" in locals():')
lines.append(' response_details.update({')
lines.append(' "status_code": getattr(response, "status_code", None),')
lines.append(' "headers": dict(getattr(response, "headers", {})),')
lines.append(' "body": getattr(response, "text", "")[:1000] if getattr(response, "text", "") else "",')
lines.append(' "elapsed": elapsed_time')
lines.append(' })')
lines.append(' raise')
lines.append('')
# Store detailed results
lines.append(' # Store detailed results')
lines.append(' result_entry = {')
lines.append(' "step": %d,' % step_num)
lines.append(' "status": response.status_code if "response" in locals() else None,')
lines.append(' "method": "%s",' % method)
lines.append(' "path": "%s",' % path)
lines.append(' "elapsed_time": elapsed_time,')
lines.append(' "request": request_details,')
lines.append(' "response": response_details,')
lines.append(' "error": error_details')
lines.append(' }')
lines.append(' self.results.append(result_entry)')
lines.append('')
# Summary methods
lines.append(' def print_summary(self):')
lines.append(' print("\\n" + "="*60)')
lines.append(' print("EXECUTION SUMMARY")')
lines.append(' print("="*60)')
lines.append(' for r in self.results:')
lines.append(' print(f"✓ Step {r['step']}: {r['method']} {r['path']} - {r['status']} ({r['elapsed_time']:.3f}s)")')
lines.append(' print("="*60)')
lines.append('')
lines.append(' def print_detailed_report(self):')
lines.append(' """Print detailed execution report with metrics"""')
lines.append(' print("\\n" + "="*80)')
lines.append(' print("DETAILED EXECUTION REPORT")')
lines.append(' print("="*80)')
lines.append(' ')
lines.append(' total_time = 0')
lines.append(' successful_steps = 0')
lines.append(' failed_steps = 0')
lines.append(' ')
lines.append(' for r in self.results:')
lines.append(' print(f"\\n--- Step {r['step']}: {r['method']} {r['path']} ---")')
lines.append(' print(f" Status: {r['status']}")')
lines.append(' print(f" Elapsed Time: {r['elapsed_time']:.3f}s")')
lines.append(' ')
lines.append(' if r['error'] is not None:')
lines.append(' print(f" Error: {r['error']}")')
lines.append(' failed_steps += 1')
lines.append(' else:')
lines.append(' successful_steps += 1')
lines.append(' ')
lines.append(' # Request details')
lines.append(' req = r['request']')
lines.append(' if req['payload'] is not None:')
lines.append(' print(f" Request Payload: {req['payload']}")')
lines.append(' ')
lines.append(' # Response details')
lines.append(' resp = r['response']')
lines.append(' if resp['headers'] is not None:')
lines.append(' content_type = resp['headers'].get('Content-Type', 'Unknown')')
lines.append(' print(f" Content-Type: {content_type}")')
lines.append(' ')
lines.append(' total_time += r['elapsed_time']')
lines.append(' ')
lines.append(' print("\\n" + "-"*80)')
lines.append(' print("SUMMARY STATISTICS")')
lines.append(' print("-"*80)')
lines.append(' print(f" Total Steps: {len(self.results)}")')
lines.append(' print(f" Successful: {successful_steps}")')
lines.append(' print(f" Failed: {failed_steps}")')
lines.append(' print(f" Total Execution Time: {total_time:.3f}s")')
lines.append(' if len(self.results) > 0:')
lines.append(' avg_time = total_time / len(self.results)')
lines.append(' print(f" Average Time per Step: {avg_time:.3f}s")')
lines.append(' print("="*80)')
lines.append('')
# Main
lines.append('def main():')
lines.append(f' BASE_URL = "{base_url}"')
lines.append(f' AUTH_HEADERS = {json.dumps(auth_headers)}')
lines.append(' executor = APIFlowExecutor(BASE_URL, AUTH_HEADERS)')
lines.append(' success = executor.execute_flow()')
lines.append(' executor.print_summary()')
lines.append(' # Check if DETAILED_REPORT environment variable is set')
lines.append(' import os')
lines.append(' if os.environ.get("DETAILED_REPORT", "").lower() == "true":')
lines.append(' executor.print_detailed_report()')
lines.append(' sys.exit(0 if success else 1)')
lines.append('')
lines.append('if __name__ == "__main__":')
lines.append(' main()')
return '\n'.join(lines)
# Example usage:
# script = generate_python_script(dependency_analysis['execution_order'], base_url, auth_headers)
# script = generate_python_script(dependency_analysis['execution_order'], base_url, auth_headers, parallel_execution=True, parallel_groups=parallel_groups)
Phase 5: Execute and Iterate
Execute this code to run the script and fix errors:
import subprocess
import tempfile
import os
import re
def execute_script_with_retries(script_content: str, max_retries: int = 5, detailed_reporting: bool = False):
"""Execute script and retry with fixes"""
for attempt in range(1, max_retries + 1):
print(f"\n=== Attempt {attempt}/{max_retries} ===")
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
f.write(script_content)
script_path = f.name
try:
# Set environment for detailed reporting if requested
env = os.environ.copy()
if detailed_reporting:
env["DETAILED_REPORT"] = "true"
result = subprocess.run(
['python', script_path],
capture_output=True,
text=True,
timeout=300,
env=env
)
print(result.stdout)
if result.returncode == 0:
print("\n✓ SUCCESS! All requests returned 2xx")
return {
'success': True,
'script': script_content,
'attempts': attempt
}
# Analyze errors and apply fixes
print(f"✗ Exit code: {result.returncode}")
# Simple fix patterns
if '400' in result.stdout and 'missing required field' in result.stdout:
# Add missing fields
field_match = re.search(r"field '(\w+)'", result.stdout)
if field_match:
field = field_match.group(1)
script_content = script_content.replace(
'payload = {',
f'payload = {{"{field}": "test_value", '
)
print(f"Applied fix: Added missing field '{field}'")
continue
if '422' in result.stdout:
# Adjust constraint violations
script_content = script_content.replace('"quantity": 0', '"quantity": 1')
script_content = script_content.replace('"age": 0', '"age": 18')
print("Applied fix: Adjusted values to meet constraints")
continue
break
except subprocess.TimeoutExpired:
print("✗ Script execution timed out")
break
except Exception as e:
print(f"✗ Execution error: {e}")
break
finally:
if os.path.exists(script_path):
os.unlink(script_path)
return {
'success': False,
'script': script_content,
'attempts': max_retries
}
# Example usage:
# result = execute_script_with_retries(generated_script)
# result = execute_script_with_retries(generated_script, detailed_reporting=True)
Complete End-to-End Example
Here's how to execute the entire workflow:
# 1. Setup
auth_headers = setup_authentication("bearer", {"token": "YOUR_TOKEN"})
# 2. Parse specification (auto-detects OpenAPI/GraphQL)
parsed_spec = parse_specification("https://api.example.com/openapi.json")
print(f"Found {len(parsed_spec['endpoints'])} endpoints")
# 3. Analyze dependencies
dependency_analysis = analyze_dependencies(parsed_spec['endpoints'])
parallel_groups = identify_parallel_groups(dependency_analysis['execution_order'])
print(f"Execution order: {len(dependency_analysis['execution_order'])} steps")
# 4. Generate script with enhanced features
generated_script = generate_python_script(
dependency_analysis['execution_order'],
parsed_spec['base_url'],
auth_headers,
parallel_execution=True, # Enable parallel execution
parallel_groups=parallel_groups
)
print(f"Generated script: {len(generated_script)} characters")
# 5. Execute with retries and detailed reporting
final_result = execute_script_with_retries(generated_script, max_retries=5, detailed_reporting=True)
# 6. Output results
if final_result['success']:
print("\n" + "="*60)
print("✓ HAPPYFLOW SCRIPT GENERATED SUCCESSFULLY")
print("="*60)
print(f"Attempts required: {final_result['attempts']}")
print("\nFinal Script:")
print(final_result['script'])
else:
print("\n✗ Failed to generate working script")
print("Manual intervention required")
Usage Instructions
When invoked, execute this skill by:
- Receive input from user (API spec URL + credentials)
- Execute Phase 1 code with user's auth credentials
- Execute Phase 2 code with spec URL
- Execute Phase 3 code with parsed endpoints
- Execute Phase 4 code to generate script with enhanced features
- Execute Phase 5 code to test and fix script
- Return final working script to user
Output Format
Return to use
…(truncated)