Session Management

Master browser session configuration and run multiple workflows in parallel for maximum efficiency.

Session Configuration Basics

Create sessions with custom settings:

from dari import Dari

client = Dari(api_key="YOUR_API_KEY")

# Basic session
session = client.create_session(
    screen_config={"width": 1440, "height": 900},
    ttl=3600  # 1 hour in seconds
)

session_id = session["session_id"]

try:
    # Your workflow here
    pass

finally:
    client.terminate_session(session_id)

Screen Configurations

Set different viewport sizes for different use cases:

from dari import Dari

client = Dari(api_key="YOUR_API_KEY")

# Desktop (standard)
desktop_session = client.create_session(
    screen_config={"width": 1920, "height": 1080},
    ttl=3600
)

# Desktop (laptop)
laptop_session = client.create_session(
    screen_config={"width": 1440, "height": 900},
    ttl=3600
)

# Tablet (iPad)
tablet_session = client.create_session(
    screen_config={"width": 768, "height": 1024},
    ttl=3600
)

# Mobile (iPhone)
mobile_session = client.create_session(
    screen_config={"width": 375, "height": 667},
    ttl=3600
)

# Mobile (Android)
android_session = client.create_session(
    screen_config={"width": 360, "height": 640},
    ttl=3600
)

try:
    # Use different sessions for different viewport tests
    result = client.run_single_action(
        action="Navigate to example.com",
        session_id=mobile_session["session_id"],
        id="mobile-nav"
    )
    print("Tested mobile view")

finally:
    # Clean up all sessions
    client.terminate_session(desktop_session["session_id"])
    client.terminate_session(laptop_session["session_id"])
    client.terminate_session(tablet_session["session_id"])
    client.terminate_session(mobile_session["session_id"])
    client.terminate_session(android_session["session_id"])

TTL (Time to Live)

Set appropriate session lifetimes:

from dari import Dari

client = Dari(api_key="YOUR_API_KEY")

# Short workflow (5 minutes)
quick_session = client.create_session(
    screen_config={"width": 1440, "height": 900},
    ttl=300  # 5 minutes
)

# Medium workflow (30 minutes)
medium_session = client.create_session(
    screen_config={"width": 1440, "height": 900},
    ttl=1800  # 30 minutes
)

# Long workflow (1 hour)
long_session = client.create_session(
    screen_config={"width": 1440, "height": 900},
    ttl=3600  # 1 hour
)

# Maximum (2 hours)
max_session = client.create_session(
    screen_config={"width": 1440, "height": 900},
    ttl=7200  # 2 hours
)

Sessions automatically terminate after TTL expires. Choose TTL based on expected workflow duration plus buffer time.

Proxy Configuration

Use proxies for location-specific testing:

from dari import Dari

client = Dari(api_key="YOUR_API_KEY")

# With residential proxy in specific city
proxied_session = client.create_session(
    screen_config={"width": 1440, "height": 900},
    ttl=3600,
    use_proxy=True,
    proxy_city="London"  # Or "New York", "Tokyo", etc.
)

try:
    # This request appears to come from London
    result = client.run_single_action(
        action="Navigate to example.com and check location",
        session_id=proxied_session["session_id"],
        id="check-location"
    )
    print(f"Result: {result['result']}")

finally:
    client.terminate_session(proxied_session["session_id"])

Custom User Agents

Set user agents per action:

from dari import Dari

client = Dari(api_key="YOUR_API_KEY")

session = client.create_session(
    screen_config={"width": 1440, "height": 900},
    ttl=3600
)

try:
    # Desktop Chrome
    desktop_result = client.run_single_action(
        action="Navigate to the site",
        session_id=session["session_id"],
        id="desktop-nav",
        user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
    )

    # Mobile Safari
    mobile_result = client.run_single_action(
        action="Check mobile version",
        session_id=session["session_id"],
        id="mobile-check",
        user_agent="Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.0 Mobile/15E148 Safari/604.1"
    )

finally:
    client.terminate_session(session["session_id"])

Parallel Workflows

Run multiple independent workflows concurrently:

from dari import Dari
import concurrent.futures

client = Dari(api_key="YOUR_API_KEY")

def process_user(user_email):
    """Process a single user workflow"""
    session = client.create_session(
        screen_config={"width": 1440, "height": 900},
        ttl=3600
    )

    try:
        # Perform workflow for this user
        result = client.run_single_action(
            action="Send onboarding email to {{email}}",
            session_id=session["session_id"],
            id=f"onboard-{user_email}",
            variables={"email": user_email}
        )

        return {
            "email": user_email,
            "success": True,
            "result": result
        }

    except Exception as e:
        return {
            "email": user_email,
            "success": False,
            "error": str(e)
        }

    finally:
        client.terminate_session(session["session_id"])

# Process multiple users in parallel
users = [
    "[email protected]",
    "[email protected]",
    "[email protected]",
    "[email protected]",
    "[email protected]"
]

# Use ThreadPoolExecutor for concurrent processing
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    results = list(executor.map(process_user, users))

# Review results
for result in results:
    if result["success"]:
        print(f"✓ {result['email']}: Success")
    else:
        print(f"✗ {result['email']}: {result['error']}")

Parallel with Different Configurations

Run parallel workflows with different session configs:

from dari import Dari
import concurrent.futures

client = Dari(api_key="YOUR_API_KEY")

def test_viewport(viewport_name, width, height):
    """Test site on specific viewport"""
    session = client.create_session(
        screen_config={"width": width, "height": height},
        ttl=1800
    )

    try:
        # Navigate and test
        nav_result = client.run_single_action(
            action="Navigate to example.com",
            session_id=session["session_id"],
            id=f"{viewport_name}-nav"
        )

        # Check layout
        check_result = client.run_single_action(
            action="Check if the navigation menu is visible",
            session_id=session["session_id"],
            id=f"{viewport_name}-check"
        )

        return {
            "viewport": viewport_name,
            "success": True,
            "menu_visible": "visible" in check_result['result'].lower()
        }

    except Exception as e:
        return {
            "viewport": viewport_name,
            "success": False,
            "error": str(e)
        }

    finally:
        client.terminate_session(session["session_id"])

# Define viewports to test
viewports = [
    ("Desktop", 1920, 1080),
    ("Laptop", 1440, 900),
    ("Tablet", 768, 1024),
    ("Mobile", 375, 667)
]

# Test all viewports in parallel
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
    futures = [
        executor.submit(test_viewport, name, width, height)
        for name, width, height in viewports
    ]
    results = [f.result() for f in concurrent.futures.as_completed(futures)]

# Display results
print("\nViewport Test Results:")
for result in results:
    if result["success"]:
        status = "✓" if result["menu_visible"] else "✗"
        print(f"{status} {result['viewport']}: Menu {'visible' if result['menu_visible'] else 'hidden'}")
    else:
        print(f"✗ {result['viewport']}: Error - {result['error']}")

Session Pool Pattern

Reuse sessions efficiently:

from dari import Dari
from queue import Queue
import threading

client = Dari(api_key="YOUR_API_KEY")

class SessionPool:
    def __init__(self, size=5):
        self.pool = Queue(maxsize=size)
        self.sessions = []

        # Create pool of sessions
        for _ in range(size):
            session = client.create_session(
                screen_config={"width": 1440, "height": 900},
                ttl=3600
            )
            self.sessions.append(session["session_id"])
            self.pool.put(session["session_id"])

    def acquire(self):
        """Get a session from the pool"""
        return self.pool.get()

    def release(self, session_id):
        """Return a session to the pool"""
        self.pool.put(session_id)

    def cleanup(self):
        """Terminate all sessions"""
        for session_id in self.sessions:
            try:
                client.terminate_session(session_id)
            except:
                pass

# Usage
pool = SessionPool(size=3)

def process_task(task_id):
    session_id = pool.acquire()
    try:
        result = client.run_single_action(
            action=f"Process task {task_id}",
            session_id=session_id,
            id=f"task-{task_id}"
        )
        print(f"Task {task_id} completed")
        return result
    finally:
        pool.release(session_id)

try:
    # Process tasks using the pool
    tasks = range(1, 11)  # 10 tasks
    for task_id in tasks:
        process_task(task_id)

finally:
    pool.cleanup()

Session State Management

Track and manage session state:

from dari import Dari
from dataclasses import dataclass
from typing import Optional
from datetime import datetime

client = Dari(api_key="YOUR_API_KEY")

@dataclass
class SessionState:
    session_id: str
    created_at: datetime
    last_action: Optional[str] = None
    action_count: int = 0
    is_logged_in: bool = False

class SessionManager:
    def __init__(self):
        self.sessions = {}

    def create_session(self, name: str):
        """Create and track a new session"""
        session = client.create_session(
            screen_config={"width": 1440, "height": 900},
            ttl=3600
        )

        state = SessionState(
            session_id=session["session_id"],
            created_at=datetime.now()
        )

        self.sessions[name] = state
        return state

    def run_action(self, name: str, action: str, **kwargs):
        """Run action and update state"""
        state = self.sessions[name]

        result = client.run_single_action(
            action=action,
            session_id=state.session_id,
            **kwargs
        )

        # Update state
        state.last_action = action
        state.action_count += 1

        # Track login status
        if "login" in action.lower() or "sign in" in action.lower():
            state.is_logged_in = True

        return result

    def get_state(self, name: str):
        """Get current session state"""
        return self.sessions[name]

    def cleanup_all(self):
        """Terminate all managed sessions"""
        for state in self.sessions.values():
            try:
                client.terminate_session(state.session_id)
            except:
                pass

# Usage
manager = SessionManager()

try:
    # Create multiple managed sessions
    manager.create_session("user1")
    manager.create_session("user2")

    # Run actions
    manager.run_action(
        "user1",
        "Navigate to example.com",
        id="nav-1"
    )

    manager.run_action(
        "user2",
        "Navigate to example.com",
        id="nav-2"
    )

    # Check state
    state1 = manager.get_state("user1")
    print(f"User1: {state1.action_count} actions, logged in: {state1.is_logged_in}")

finally:
    manager.cleanup_all()

Best Practices

Match screen size to use case

Use appropriate viewport sizes:

Desktop testing: 1920x1080 or 1440x900
Mobile testing: 375x667 (iPhone) or 360x640 (Android)
Tablet testing: 768x1024

Set realistic TTL

Add buffer time to expected workflow duration:

estimated_duration = 1200  # 20 minutes
buffer = 600  # 10 minutes
ttl = estimated_duration + buffer  # 30 minutes

Limit parallel sessions

Don’t create too many concurrent sessions:

# Good: Reasonable concurrency
max_workers=5

# Bad: Too many sessions
max_workers=100

Always clean up sessions

Terminate sessions even in parallel workflows:

try:
    # Your workflow
    pass
finally:
    client.terminate_session(session_id)

Handle session expiration

Check TTL and handle expiration gracefully:

if session_expired:
    # Create new session
    new_session = client.create_session(...)

Next Steps

Getting Started

Learn the basics of programmatic actions

Programmatic Actions Overview

Learn about the programmatic approach

Create Session API

Complete API reference for sessions

Run Single Action API

Full API documentation for single actions

Getting Started

Visual Workflow Builder

Programmatic Actions

​Session Configuration Basics

​Screen Configurations

​TTL (Time to Live)

​Proxy Configuration

​Custom User Agents

​Parallel Workflows

​Parallel with Different Configurations

​Session Pool Pattern

​Session State Management

​Best Practices

​Next Steps

Getting Started

Programmatic Actions Overview

Create Session API

Run Single Action API

Session Configuration Basics

Screen Configurations

TTL (Time to Live)

Proxy Configuration

Custom User Agents

Parallel Workflows

Parallel with Different Configurations

Session Pool Pattern

Session State Management

Best Practices

Next Steps