Skip to main content

What is the Code Block?

The Code block lets you write Python code to manipulate workflow data. Use it when you need custom logic, complex transformations, or data processing that goes beyond what other blocks can do. Common use cases:
  • Transform and filter data
  • Parse and extract information from text
  • Perform calculations and aggregations
  • Combine data from multiple steps
  • Format data for specific outputs
  • Create custom data structures
  • Clean and validate data

How Does the Code Block Work?

  1. Write Python code directly in the block
  2. Access previous steps using step_1, step_2, etc.
  3. Process the data using Python and available libraries
  4. Return the result using return statement
  5. Output becomes available to subsequent steps

Configuration

Language

Currently supports: Python

Code (Required)

Write your Python code in this field. You don’t need to define a function - just write the code and use return to output the result. Basic example:
# Access previous step data
keyword = step_1['output']['Keyword']

# Process it
result = keyword.upper()

# Return the result
return result

Available Libraries

The Code block includes these Python libraries:
  • BeautifulSoup - HTML/XML parsing
  • numpy (np) - Numerical computations
  • pandas (pd) - Data analysis and manipulation
  • json - JSON encoding/decoding
  • datetime - Date and time handling
  • math - Mathematical functions
  • random - Random number generation
  • re - Regular expressions
  • collections - Container data types
  • markdown - Markdown processing
  • lxml - XML processing

Accessing Workflow Data

Previous Steps

Access output from previous steps using step_X notation:
# Access step output
data = step_1['output']

# Access nested properties
keyword = step_1['output']['Keyword']
volume = step_1['output']['Search_Volume']

return keyword

Input Variables

Access workflow inputs:
# Access input variables
topic = input['topic']
count = input['count']

return topic

Use Cases and Examples

Use Case 1: Extract Keywords from Array

Input from Related Keywords block:
[
  {"Keyword": "crm", "Search Volume": "165000"},
  {"Keyword": "sales", "Search Volume": "74000"},
  {"Keyword": "marketing", "Search Volume": "33100"}
]
Code:
# Get keywords from step 1
keywords_data = step_1['output']

# Extract just the keyword names
keywords = [item['Keyword'] for item in keywords_data]

return keywords
Output:
["crm", "sales", "marketing"]

Use Case 2: Filter High-Volume Keywords

Input:
[
  {"Keyword": "crm", "Search Volume": "165000"},
  {"Keyword": "email", "Search Volume": "5400"},
  {"Keyword": "marketing", "Search Volume": "74000"}
]
Code:
# Get keywords
keywords = step_1['output']

# Filter keywords with volume greater than 10000
high_volume = [
    kw for kw in keywords
    if int(kw['Search Volume']) > 10000
]

return high_volume
Output:
[
  {"Keyword": "crm", "Search Volume": "165000"},
  {"Keyword": "marketing", "Search Volume": "74000"}
]

Use Case 3: Calculate Statistics

Input:
[
  {"Keyword": "crm", "Search Volume": "165000"},
  {"Keyword": "sales", "Search Volume": "74000"},
  {"Keyword": "marketing", "Search Volume": "33100"}
]
Code:
# Get keywords
keywords = step_1['output']

# Calculate statistics
volumes = [int(kw['Search Volume']) for kw in keywords]
total_volume = sum(volumes)
avg_volume = total_volume / len(volumes)
max_volume = max(volumes)
min_volume = min(volumes)

# Return summary
return {
    'total_volume': total_volume,
    'average_volume': round(avg_volume, 2),
    'max_volume': max_volume,
    'min_volume': min_volume,
    'keyword_count': len(keywords)
}
Output:
{
  "total_volume": 272100,
  "average_volume": 90700.0,
  "max_volume": 165000,
  "min_volume": 33100,
  "keyword_count": 3
}

Use Case 4: Parse and Extract from Text

Input: Text with URLs Code:
import re

# Get text from previous step
text = step_1['output']

# Extract all URLs using regex
url_pattern = r'https?://[^\s<>"{}|\\^`\[\]]+'
urls = re.findall(url_pattern, text)

# Remove duplicates
unique_urls = list(set(urls))

return unique_urls

Use Case 5: Create Structured Data

Input: Array of keywords Code:
# Get keywords
keywords = step_1['output']

# Create structured data with categorization
structured = []

for kw in keywords:
    volume = int(kw['Search Volume'])

    # Categorize by volume
    if volume > 100000:
        category = 'high'
        priority = 1
    elif volume > 10000:
        category = 'medium'
        priority = 2
    else:
        category = 'low'
        priority = 3

    structured.append({
        'keyword': kw['Keyword'],
        'volume': volume,
        'category': category,
        'priority': priority
    })

# Sort by priority
structured.sort(key=lambda x: x['priority'])

return structured

Use Case 6: Format Data for Export

Input: Keyword data Code:
# Get keywords
keywords = step_1['output']

# Format as CSV-ready structure
csv_data = []

# Add header
csv_data.append(['Keyword', 'Search Volume', 'Status'])

# Add rows
for kw in keywords:
    volume = int(kw['Search Volume'])
    status = 'High Priority' if volume > 50000 else 'Standard'
    csv_data.append([kw['Keyword'], volume, status])

return csv_data
Output:
[
  ["Keyword", "Search Volume", "Status"],
  ["crm", 165000, "High Priority"],
  ["sales", 74000, "High Priority"],
  ["marketing", 33100, "Standard"]
]

Use Case 7: Clean and Normalize Data

Input: Data with inconsistent formatting Code:
# Get data
data = step_1['output']

# Clean and normalize
cleaned = []

for item in data:
    cleaned_item = {
        # Normalize keyword to lowercase
        'keyword': item['Keyword'].lower().strip(),

        # Convert volume to integer
        'volume': int(item['Search Volume'].replace(',', '')),

        # Ensure CPC is float
        'cpc': float(item.get('CPC', '0'))
    }

    cleaned.append(cleaned_item)

return cleaned

Use Case 8: Group and Aggregate Data

Input: Keywords with categories Code:
from collections import defaultdict

# Get keywords
keywords = step_1['output']

# Group by difficulty range
groups = defaultdict(list)

for kw in keywords:
    difficulty = int(kw['Keyword Difficulty Index'])

    if difficulty >= 70:
        group = 'very_difficult'
    elif difficulty >= 50:
        group = 'difficult'
    elif difficulty >= 30:
        group = 'moderate'
    else:
        group = 'easy'

    groups[group].append(kw['Keyword'])

# Convert to regular dict
result = {
    'very_difficult': groups['very_difficult'],
    'difficult': groups['difficult'],
    'moderate': groups['moderate'],
    'easy': groups['easy'],
    'summary': {
        'very_difficult_count': len(groups['very_difficult']),
        'difficult_count': len(groups['difficult']),
        'moderate_count': len(groups['moderate']),
        'easy_count': len(groups['easy'])
    }
}

return result

Use Case 9: Parse HTML Content

Input: HTML content from web scraping Code:
from bs4 import BeautifulSoup

# Get HTML content
html = step_1['output']

# Parse HTML
soup = BeautifulSoup(html, 'html.parser')

# Extract specific elements
title = soup.find('title').text if soup.find('title') else 'No title'
headings = [h.text.strip() for h in soup.find_all(['h1', 'h2', 'h3'])]
paragraphs = [p.text.strip() for p in soup.find_all('p') if p.text.strip()]

# Return structured data
return {
    'title': title,
    'headings': headings,
    'paragraph_count': len(paragraphs),
    'first_paragraph': paragraphs[0] if paragraphs else None
}

Use Case 10: Generate Date Ranges

Code:
import datetime

# Get date parameters from input
start_date = datetime.datetime.strptime(input['start_date'], '%Y-%m-%d')
end_date = datetime.datetime.strptime(input['end_date'], '%Y-%m-%d')

# Generate list of dates
date_list = []
current = start_date

while current <= end_date:
    date_list.append(current.strftime('%Y-%m-%d'))
    current += datetime.timedelta(days=1)

return date_list

Use Case 11: Sort and Rank Data

Input: Keywords array Code:
# Get keywords
keywords = step_1['output']

# Sort by search volume (descending)
sorted_keywords = sorted(
    keywords,
    key=lambda x: int(x['Search Volume']),
    reverse=True
)

# Add rank
for i, kw in enumerate(sorted_keywords, 1):
    kw['rank'] = i

return sorted_keywords

Use Case 12: Using Pandas for Data Analysis

Input: Keyword data array Code:
import pandas as pd

# Get keywords
keywords = step_1['output']

# Create DataFrame
df = pd.DataFrame(keywords)

# Convert volume to numeric
df['Search Volume'] = df['Search Volume'].astype(int)

# Calculate statistics
stats = {
    'total_keywords': len(df),
    'total_volume': int(df['Search Volume'].sum()),
    'average_volume': float(df['Search Volume'].mean()),
    'median_volume': float(df['Search Volume'].median()),
    'top_keyword': df.loc[df['Search Volume'].idxmax(), 'Keyword']
}

return stats

Use Case 13: Filter by Multiple Conditions

Input: Keywords with multiple metrics Code:
# Get keywords
keywords = step_1['output']

# Filter by multiple conditions
filtered = [
    kw for kw in keywords
    if int(kw['Search Volume']) > 10000
    and int(kw['Keyword Difficulty Index']) < 50
    and float(kw['CPC']) > 5.0
]

# Sort by volume
filtered.sort(key=lambda x: int(x['Search Volume']), reverse=True)

return filtered

Use Case 14: Create Summary Report

Input: Multiple steps with different data Code:
# Get data from multiple steps
keywords = step_1['output']
domain = step_2['output']
content = step_3['output']

# Create comprehensive summary
summary = {
    'analysis_date': datetime.datetime.now().strftime('%Y-%m-%d'),
    'keywords': {
        'total_count': len(keywords),
        'high_volume_count': sum(1 for kw in keywords if int(kw['Search Volume']) > 50000),
        'total_search_volume': sum(int(kw['Search Volume']) for kw in keywords),
        'top_keyword': max(keywords, key=lambda x: int(x['Search Volume']))['Keyword']
    },
    'domain': {
        'name': domain['Domain'],
        'organic_traffic': domain['Organic Traffic'],
        'organic_keywords': domain['Organic Keywords']
    },
    'content': {
        'length': len(content.split()) if isinstance(content, str) else 0,
        'type': 'text'
    }
}

return summary

Best Practices

Code Organization

  • Keep it simple: Write clear, readable code
  • Comment complex logic: Use # for comments
  • One responsibility: Each code block should do one thing well
  • Test incrementally: Start simple, add complexity gradually

Data Handling

  • Validate input: Check if data exists before processing
  • Handle errors: Use try/except for operations that might fail
  • Convert types: Ensure data types are correct (int, float, str)
  • Return meaningful data: Structure output for downstream blocks

Performance

  • Avoid unnecessary loops: Use list comprehensions when possible
  • Filter early: Reduce data size before complex operations
  • Limit iterations: Maximum 25 loops allowed
  • Be efficient: Use built-in functions and libraries

Common Patterns

Safe data access:
# Check if key exists
data = step_1['output'] if 'output' in step_1 else []
keyword = data[0]['Keyword'] if data else 'default'

return keyword
Type conversion:
# Convert strings to numbers safely
volume = int(step_1['output']['Search Volume'].replace(',', ''))
cpc = float(step_1['output']['CPC'])

return volume
Error handling:
try:
    # Attempt operation
    result = int(step_1['output']['volume'])
except (KeyError, ValueError, TypeError):
    # Handle error
    result = 0

return result

Debugging

Using Print Statements

Use print() to log information during execution:
# Debug output
keywords = step_1['output']
print(f"Processing {len(keywords)} keywords")

for kw in keywords:
    print(f"Keyword: {kw['Keyword']}, Volume: {kw['Search Volume']}")

# Filter and return
filtered = [kw for kw in keywords if int(kw['Search Volume']) > 10000]
print(f"Filtered to {len(filtered)} keywords")

return filtered
Logs appear in the workflow execution output.

Limitations

Security Restrictions

These modules are blocked for security:
  • os - Operating system interface
  • sys - System-specific parameters
  • subprocess - Subprocess management
  • shutil - File operations
  • socket - Network connections
  • multiprocessing - Process-based parallelism

Loop Limits

Maximum of 25 loops allowed per code block execution. If you need more iterations, consider:
  • Processing data in batches with multiple code blocks
  • Using the Loop block for iteration
  • Simplifying your logic

Troubleshooting

Code Not Returning Output

Problem: Block executes but output is empty. Solution:
  • Make sure you use return statement
  • Check your return value is not None
  • Verify data structure is correct
# Wrong - no return
keywords = step_1['output']
filtered = [kw for kw in keywords if int(kw['Search Volume']) > 10000]

# Correct - has return
keywords = step_1['output']
filtered = [kw for kw in keywords if int(kw['Search Volume']) > 10000]
return filtered

KeyError: Step or Property Not Found

Problem: Error accessing step data. Solution:
  • Check step number is correct
  • Verify previous step completed successfully
  • Use safe access patterns
# Unsafe
keyword = step_1['output']['Keyword']

# Safe
output = step_1.get('output', {})
keyword = output.get('Keyword', 'default')
return keyword

Type Errors

Problem: Operations fail due to wrong data types. Solution:
  • Convert strings to numbers explicitly
  • Check data types before operations
  • Handle missing or null values
# Convert string to int
volume = int(step_1['output']['Search Volume'])

# Handle potential errors
try:
    volume = int(step_1['output']['Search Volume'])
except (ValueError, KeyError, TypeError):
    volume = 0

return volume

Loop Limit Exceeded

Problem: Code exceeds 25 loop limit. Solution:
  • Reduce iterations
  • Use list comprehensions (more efficient)
  • Use Loop block for large iterations
# Instead of nested loops, use comprehensions
result = [
    {'keyword': kw['Keyword'], 'priority': calc_priority(kw)}
    for kw in step_1['output']
]

return result

Error Handling

The Code block will fail if:
  • Python syntax is incorrect
  • Referenced data doesn’t exist
  • Loop limit is exceeded
  • Blocked modules are imported
  • Execution times out
Set error handling to determine workflow behavior on failure:
  • Terminate Workflow: Stop execution
  • Continue Execution: Proceed with None output

What’s Next

Now that you understand the Code block: