Code Generation

from workflowapp.library.base_block import BaseBlock

class GeneratedBlock(BaseBlock):
    def execute(self, inputs, config):
        # Generated code here
        ...

from library.social_scraping_library import SocialScrapingLibrary
from library.llm_ops_library import LLMOpsLibrary

# Use tools
scraper = SocialScrapingLibrary()
llm = LLMOpsLibrary()

def execute(self, inputs, config):
    # Get data from previous block
    previous_output = inputs.get("previous_block", {})
    posts = previous_output.get("posts", [])
    
    # Process data
    filtered_posts = [p for p in posts if p["upvotes"] > 50]
    
    # Return output
    return {"filtered_posts": filtered_posts}

Filter Reddit posts with 50+ upvotes from the last 24 hours

from workflowapp.library.base_block import BaseBlock
from datetime import datetime, timedelta

class FilterPostsBlock(BaseBlock):
    def execute(self, inputs, config):
        posts = inputs.get("posts", [])
        min_upvotes = config.get("min_upvotes", 50)
        
        # Filter by upvotes
        filtered = [p for p in posts if p.get("upvotes", 0) >= min_upvotes]
        
        # Filter by time (last 24 hours)
        cutoff = datetime.now() - timedelta(hours=24)
        recent = [p for p in filtered if p.get("created_at") > cutoff]
        
        return {"filtered_posts": recent}

Scrape Reddit posts from r/python and send email summary

from workflowapp.library.base_block import BaseBlock
from library.social_scraping_library import SocialScrapingLibrary
from library.communication_tools_library import CommunicationToolsLibrary

class ScrapeAndEmailBlock(BaseBlock):
    def execute(self, inputs, config):
        scraper = SocialScrapingLibrary()
        email_tool = CommunicationToolsLibrary()
        
        # Scrape posts
        result = scraper.get_posts_best(
            subreddits=["r/python"],
            limit=10
        )
        posts = result.get("items", [])
        
        # Create summary
        summary = f"Found {len(posts)} posts from r/python"
        
        # Send email
        email_tool.send_email(
            to=config.get("notification_email"),
            subject="Reddit Summary",
            body=summary
        )
        
        return {"posts": posts, "summary": summary}

def execute(self, inputs, config):
    try:
        # Main logic
        result = some_operation()
        return {"success": True, "data": result}
    except Exception as e:
        rilo_logger.error(f"Error in block: {str(e)}")
        return {"success": False, "error": str(e)}