eywa-client 0.4.2

EYWA client library for Python providing JSON-RPC communication, GraphQL queries, and task management for EYWA robots

EYWA Client for Python

MODERNIZED EYWA client library for Python providing JSON-RPC communication, GraphQL queries, and comprehensive file operations for EYWA robots.

🚀 Version 0.4.0 - Path-Based Operations

Breaking Change: eywa.graphql() now returns data directly instead of wrapping in {"data": ...}.

# Before (v0.3.x)
result = await eywa.graphql("{ searchUser { name } }")
users = result["data"]["searchUser"]

# After (v0.4.0)
result = await eywa.graphql("{ searchUser { name } }")
users = result["searchUser"]  # Data returned directly

New in v0.4.0

• ✅ ensure_path() - Auto-create nested folder structures by path
• ✅ get_folder_by_path() - Find folders by path string
• ✅ folder_path parameter - Upload files with auto-created folders
• ✅ Simplified GraphQL API - Returns data directly, errors raise exceptions

Existing Features

• ✅ Single Map Arguments - API functions use single dict arguments that mirror GraphQL schema
• ✅ Client UUID Management - Full control over file and folder UUIDs for deduplication
• ✅ Modern GraphQL Patterns - Relationship filtering instead of broken WHERE clause patterns
• ✅ Complete Folder Operations - Full folder hierarchy support (create, list, delete, info)
• ✅ Streaming Operations - Memory-efficient uploads/downloads with progress tracking

Installation

pip install eywa-client

Quick Start

import asyncio
import eywa
import eywa_files  # Import file operations module

async def main():
    # Initialize the client
    eywa.open_pipe()

    # Log messages
    eywa.info("Robot started")

    # Execute GraphQL queries
    result = await eywa.graphql("""
        query {
            searchUser(_limit: 10) {
                euuid
                name
                type
            }
        }
    """)

    # Upload a file (from eywa_files module)
    file_uuid = "550e8400-e29b-41d4-a716-446655440000"
    file_info = await eywa_files.upload_content("Hello from EYWA!", {
        "name": "greeting.txt",
        "euuid": file_uuid,
        "content_type": "text/plain"
    })
    eywa.info(f"Uploaded: {file_info['name']} ({file_info['euuid']})")

    # Update task status
    eywa.update_task(eywa.PROCESSING)

    # Complete the task
    eywa.close_task(eywa.SUCCESS)

asyncio.run(main())

Features

Core Features

• 🚀 Async/Await Support - Modern Python async programming
• 📡 JSON-RPC Communication - Seamless communication with EYWA server
• 🗃️ GraphQL Integration - Execute queries and mutations
• 📝 Task Management - Status updates, logging, and reporting

File Operations (NEW - v2.0)

• 📤 Modern Upload API - Single map arguments matching GraphQL schema
• 📥 Streaming Downloads - Memory-efficient downloads with progress tracking
• 📁 Complete Folder Support - Create, list, delete, and manage folder hierarchies
• 🔧 Client UUID Control - Pre-generate UUIDs for deterministic file management
• 🚀 S3 Integration - 3-step upload protocol (request → upload → confirm)
• 📊 Progress Callbacks - Track upload/download progress in real-time
• 📊 GraphQL Integration - Execute queries and mutations against EYWA datasets
• 📝 Comprehensive Logging - Multiple log levels with metadata support
• 🔄 Task Management - Update status, report progress, handle task lifecycle
• 🎯 Type Hints - Full type annotations for better IDE support
• 📋 Table/Sheet Classes - Built-in data structures for reports

API Reference

Initialization

open_pipe()

Initialize stdin/stdout communication with EYWA runtime. Must be called before using other functions.

eywa.open_pipe()

Logging Functions

log(event="INFO", message="", data=None, duration=None, coordinates=None, time=None)

Log a message with full control over all parameters.

eywa.log(
    event="INFO",
    message="Processing item",
    data={"itemId": 123},
    duration=1500,
    coordinates={"x": 10, "y": 20}
)

info(), error(), warn(), debug(), trace(), exception()

Convenience methods for different log levels.

eywa.info("User logged in", {"userId": "abc123"})
eywa.error("Failed to process", {"error": str(e)})
eywa.exception("Unhandled error", {"stack": traceback.format_exc()})

Task Management

async get_task()

Get current task information. Returns a coroutine.

task = await eywa.get_task()
print(f"Processing: {task['message']}")

update_task(status="PROCESSING")

Update the current task status.

eywa.update_task(eywa.PROCESSING)

close_task(status="SUCCESS")

Close the task with a final status and exit the process.

try:
    # Do work...
    eywa.close_task(eywa.SUCCESS)
except Exception as e:
    eywa.error("Task failed", {"error": str(e)})
    eywa.close_task(eywa.ERROR)

return_task()

Return control to EYWA without closing the task.

eywa.return_task()

📁 File Operations (v2.0 - Modernized)

The modernized Python client provides comprehensive file and folder operations via the eywa_files module.

Import File Operations

import eywa
import eywa_files  # Separate module for file operations

# Or import specific functions
from eywa_files import (
    upload,
    upload_content,
    upload_stream,
    download,
    download_stream,
    create_folder,
    delete_file,
    delete_folder,
    ensure_path,
    get_folder_by_path,
    ROOT_UUID,
    ROOT_FOLDER,
    FileUploadError,
    FileDownloadError
)

Key Concepts

• Single Map Arguments - All functions use single dict arguments that mirror GraphQL schema
• Client UUID Control - You generate and manage UUIDs for deterministic operations
• Path-Based Operations - Use folder_path or ensure_path() for intuitive folder management
• 3-Step Upload Protocol - Request URL → S3 Upload → Confirm
• Complete Folder Support - Full hierarchy management

Best Practices

File UUID Management

When uploading to a folder (folder or folder_path specified):

The service automatically checks if a file with the same name already exists in that folder:

• If the file exists → Reuses the existing file's UUID (overwrites the file)
• If the file doesn't exist → Creates a new file with auto-generated UUID

# DON'T specify euuid when uploading to folders
# Let the service manage UUIDs based on filename
await upload("report.pdf", {
    "name": "monthly-report.pdf",
    "folder_path": "/reports/2024/"  # Service checks for existing file by name
})

# First upload: Creates new file with auto-generated UUID
# Second upload: Finds existing "monthly-report.pdf" and overwrites it

Only specify euuid for orphaned files:

Orphaned files are files linked to other EYWA records (e.g., user avatars, document attachments) that exist outside of folder hierarchies:

# DO specify euuid for files linked to other entities
user_avatar_uuid = user_record["avatar_file_id"]  # UUID from your data model

await upload_content(avatar_bytes, {
    "name": "avatar.jpg",
    "euuid": user_avatar_uuid,  # Controlled UUID for entity reference
    "content_type": "image/jpeg"
    # No folder - orphaned file linked to user record
})

Why this matters:

• Folders: Files are identified by name within their folder (like a filesystem)
• Orphaned files: Files are identified by UUID for database relationships
• Idempotent uploads: Uploading the same filename to the same folder multiple times safely updates the file

Example - Folder-based workflow:

# Upload daily report - no UUID needed
report_info = await upload("daily-data.csv", {
    "name": "daily-data.csv",
    "folder_path": "/reports/daily/"
})
# Returns: existing file UUID if file exists, new UUID if not

# Later upload with same name - automatically overwrites
updated_info = await upload("new-daily-data.csv", {
    "name": "daily-data.csv",  # Same name
    "folder_path": "/reports/daily/"  # Same folder
})
# Returns: same UUID, updated content

Example - Entity-linked workflow:

# File linked to a specific invoice record
invoice_id = "550e8400-e29b-41d4-a716-446655440000"
invoice_file_uuid = f"{invoice_id}-attachment"

file_info = await upload("invoice.pdf", {
    "name": "invoice-12345.pdf",
    "euuid": invoice_file_uuid  # Explicit UUID for database relationship
})

# Store reference in your invoice record
await eywa.graphql("""
    mutation UpdateInvoice($id: UUID!, $fileId: UUID!) {
        syncInvoice(data: {euuid: $id, attachment_file: $fileId}) {
            euuid
        }
    }
""", {"id": invoice_id, "fileId": invoice_file_uuid})

Path-Based Operations (NEW in v0.4.0)

Upload with folder_path

from eywa_files import upload, upload_content

# Upload file with auto-created folder structure
file_info = await upload("report.pdf", {
    "name": "report.pdf",
    "folder_path": "/projects/2024/reports/"  # Creates folders if needed
})
print(f"Uploaded to: {file_info['folder']['path']}")

# Upload content with folder_path
file_info = await upload_content(
    json.dumps({"data": "value"}),
    {
        "name": "data.json",
        "content_type": "application/json",
        "folder_path": "/exports/2024/"
    }
)
print(f"File UUID: {file_info['euuid']}, Status: {file_info['status']}")

ensure_path - Create Folder Structure

from eywa_files import ensure_path

# Create nested folders (idempotent - safe to call multiple times)
folder = await ensure_path("/projects/2024/reports/")
print(f"Folder ready: {folder['path']}")  # /projects/2024/reports/

# Returns existing folder if path exists
same_folder = await ensure_path("/projects/2024/reports/")
assert folder["euuid"] == same_folder["euuid"]

get_folder_by_path - Find Folder

from eywa_files import get_folder_by_path

# Find folder by path
folder = await get_folder_by_path("/projects/2024/")
if folder:
    print(f"Found: {folder['name']} ({folder['euuid']})")
else:
    print("Folder not found")

Constants

# Root folder for file operations
print(eywa_files.ROOT_UUID)    # "87ce50d8-5dfa-4008-a265-053e727ab793"
print(eywa_files.ROOT_FOLDER)  # {"euuid": "87ce50d8-5dfa-4008-a265-053e727ab793"}

Upload Operations

Upload Content from Memory

import uuid
from eywa_files import upload_content

# Upload string content with client UUID
file_uuid = str(uuid.uuid4())

file_info = await upload_content("Hello EYWA!", {
    "name": "greeting.txt",
    "euuid": file_uuid,  # Client controls UUID
    "folder": {"euuid": folder_uuid},
    "content_type": "text/plain"
})
print(f"Uploaded: {file_info['name']}, Status: {file_info['status']}")

# Upload JSON data
import json
data = {"message": "Hello", "timestamp": "2024-01-01"}

file_info = await upload_content(json.dumps(data), {
    "name": "data.json",
    "euuid": str(uuid.uuid4()),
    "content_type": "application/json"
})
print(f"File UUID: {file_info['euuid']}, Size: {file_info['size']} bytes")

Upload File from Disk

from eywa_files import upload

# Upload with progress tracking
def progress_callback(current, total):
    percentage = (current / total) * 100
    eywa.info(f"Upload: {percentage:.1f}% ({current}/{total} bytes)")

file_info = await upload("local_file.pdf", {
    "name": "document.pdf",
    "euuid": str(uuid.uuid4()),
    "folder": {"euuid": reports_folder_uuid},
    "progress_fn": progress_callback
})
eywa.info(f"Upload complete: {file_info['euuid']} in {file_info['folder']['path']}")

Upload from Stream

from eywa_files import upload_stream

# Upload from async iterator
async def data_generator():
    for i in range(1000):
        yield f"Line {i}\n".encode()

file_info = await upload_stream(data_generator(), {
    "name": "generated.txt",
    "size": 8000,  # Must calculate size beforehand
    "euuid": str(uuid.uuid4())
})
print(f"Stream uploaded: {file_info['name']} ({file_info['euuid']})")

Download Operations

Download to Memory

from eywa_files import download

# Download with progress tracking
def download_progress(current, total):
    eywa.info(f"Downloaded: {current}/{total} bytes")

content = await download(file_uuid, progress_fn=download_progress)
text = content.decode('utf-8')

Download to File

from eywa_files import download

# Download and save to disk
saved_path = await download(file_uuid, save_path="local_copy.txt")
eywa.info(f"File saved to: {saved_path}")

Stream Download (Memory Efficient)

from eywa_files import download_stream

# For large files - process in chunks
stream_result = await download_stream(file_uuid)

with open("large_file.dat", "wb") as f:
    async for chunk in stream_result["stream"]:
        f.write(chunk)

File Management

File management operations use direct GraphQL queries for maximum flexibility.

List Files with Modern Filtering

# List files using GraphQL directly
files_result = await eywa.graphql("""
    query ListFiles($folderUUID: UUID) {
        searchFile(_where: {
            folder: {euuid: {_eq: $folderUUID}}
        }, _order_by: {uploaded_at: desc}) {
            euuid
            name
            size
            content_type
            status
            uploaded_at
            folder {
                euuid
                name
                path
            }
        }
    }
""", {"folderUUID": folder_uuid})

files = files_result["searchFile"]

File Information

# Get detailed file info with GraphQL
file_result = await eywa.graphql("""
    query GetFile($uuid: UUID!) {
        getFile(euuid: $uuid) {
            euuid
            name
            size
            content_type
            status
            uploaded_at
            folder {
                euuid
                name
                path
            }
        }
    }
""", {"uuid": file_uuid})

file_info = file_result["getFile"]
if file_info:
    eywa.info(f"Name: {file_info['name']}")
    eywa.info(f"Size: {file_info['size']} bytes")

Delete Files

from eywa_files import delete_file

# Delete file
success = await delete_file(file_uuid)
if success:
    eywa.info("File deleted successfully")

Folder Operations

Create Folders

from eywa_files import create_folder, ROOT_UUID

# Create folder in root
folder = await create_folder({
    "name": "my-documents",
    "euuid": str(uuid.uuid4()),
    "parent": {"euuid": ROOT_UUID}
})

# Create subfolder
subfolder = await create_folder({
    "name": "reports",
    "euuid": str(uuid.uuid4()),
    "parent": {"euuid": folder["euuid"]}
})

eywa.info(f"Created: {subfolder['path']}")

List Folders

# List folders using GraphQL directly
folders_result = await eywa.graphql("""
    query ListFolders($parentUUID: UUID) {
        searchFolder(_where: {
            parent: {euuid: {_eq: $parentUUID}}
        }, _order_by: {name: asc}) {
            euuid
            name
            path
            modified_on
            parent {
                euuid
                name
            }
        }
    }
""", {"parentUUID": parent_folder_uuid})

folders = folders_result["searchFolder"]

Folder Information

# Get folder by UUID with GraphQL
folder_result = await eywa.graphql("""
    query GetFolder($uuid: UUID!) {
        getFolder(euuid: $uuid) {
            euuid
            name
            path
            modified_on
            parent {
                euuid
                name
            }
        }
    }
""", {"uuid": folder_uuid})

folder = folder_result["getFolder"]
if folder:
    eywa.info(f"Folder: {folder['name']} -> {folder['path']}")

Delete Folders

from eywa_files import delete_folder

# Delete empty folder
success = await delete_folder(folder_uuid)
if not success:
    eywa.warn("Folder deletion failed - may contain files")

Complete Example

import asyncio
import eywa
import eywa_files
from eywa_files import (
    create_folder,
    upload_content,
    download,
    ROOT_UUID,
    FileUploadError,
    FileDownloadError
)
import uuid
import json

async def file_operations_example():
    eywa.open_pipe()

    try:
        # Create folder structure
        project_uuid = str(uuid.uuid4())
        project_folder = await create_folder({
            "name": "my-project",
            "euuid": project_uuid,
            "parent": {"euuid": ROOT_UUID}
        })

        eywa.info(f"Created folder: {project_folder['path']}")

        # Upload files
        readme_uuid = str(uuid.uuid4())
        readme_info = await upload_content("# My Project\nThis is a demo", {
            "name": "README.md",
            "euuid": readme_uuid,
            "folder": {"euuid": project_uuid},
            "content_type": "text/markdown"
        })
        eywa.info(f"Uploaded README: {readme_info['euuid']}")

        # Upload JSON config
        config_data = {"version": "1.0", "debug": True}
        config_uuid = str(uuid.uuid4())
        config_info = await upload_content(json.dumps(config_data), {
            "name": "config.json",
            "euuid": config_uuid,
            "folder": {"euuid": project_uuid},
            "content_type": "application/json"
        })
        eywa.info(f"Uploaded config: {config_info['euuid']} ({config_info['size']} bytes)")

        eywa.info("Files uploaded successfully")

        # List project files using GraphQL
        files_result = await eywa.graphql("""
            query ListProjectFiles($folderUUID: UUID!) {
                searchFile(_where: {
                    folder: {euuid: {_eq: $folderUUID}}
                }, _order_by: {uploaded_at: desc}) {
                    euuid
                    name
                    size
                    content_type
                }
            }
        """, {"folderUUID": project_uuid})

        project_files = files_result["searchFile"]
        eywa.info(f"Project files ({len(project_files)}):")
        for file in project_files:
            eywa.info(f"  - {file['name']} ({file['size']} bytes)")

        # Download and verify
        config_content = await download(config_uuid)
        config_json = json.loads(config_content.decode('utf-8'))
        eywa.info(f"Config version: {config_json['version']}")

        eywa.info("File operations completed successfully")
        eywa.close_task(eywa.SUCCESS)

    except (FileUploadError, FileDownloadError) as e:
        eywa.error(f"File operation failed: {e}")
        eywa.close_task(eywa.ERROR)
    except Exception as e:
        eywa.error(f"Unexpected error: {e}")
        eywa.close_task(eywa.ERROR)

if __name__ == "__main__":
    asyncio.run(file_operations_example())

Error Handling

from eywa_files import upload_content, download, FileUploadError, FileDownloadError

try:
    await upload_content("test", {"name": "test.txt"})
except FileUploadError as e:
    eywa.error(f"Upload failed: {e}")
    # Exception has detailed information about what went wrong

try:
    content = await download("non-existent-uuid")
except FileDownloadError as e:
    eywa.error(f"Download failed: {e}")

Utility Functions

from eywa_files import calculate_file_hash

# Calculate file hash before uploading
hash_value = calculate_file_hash("local_file.txt", "sha256")
eywa.info(f"SHA256: {hash_value}")

Reporting

report(message, data=None, image=None)

Send a task report with optional data and image.

eywa.report("Analysis complete", {
    "accuracy": 0.95,
    "processed": 1000
}, chart_image_base64)

GraphQL

async graphql(query, variables=None)

Execute a GraphQL query against the EYWA server.

result = await eywa.graphql("""
    mutation CreateUser($input: UserInput!) {
        syncUser(data: $input) {
            euuid
            name
        }
    }
""", {
    "input": {
        "name": "John Doe",
        "active": True
    }
})

JSON-RPC

async send_request(data)

Send a JSON-RPC request and wait for response.

result = await eywa.send_request({
    "method": "custom.method",
    "params": {"foo": "bar"}
})

send_notification(data)

Send a JSON-RPC notification without expecting a response.

eywa.send_notification({
    "method": "custom.event",
    "params": {"status": "ready"}
})

register_handler(method, func)

Register a handler for incoming JSON-RPC method calls.

def handle_ping(data):
    print(f"Received ping: {data['params']}")
    eywa.send_notification({
        "method": "custom.pong",
        "params": {"timestamp": time.time()}
    })

eywa.register_handler("custom.ping", handle_ping)

Data Structures

Sheet Class

For creating structured tabular data:

sheet = eywa.Sheet("UserReport")
sheet.set_columns(["Name", "Email", "Status"])
sheet.add_row({"Name": "John", "Email": "[e-mail]", "Status": "Active"})
sheet.add_row({"Name": "Jane", "Email": "[e-mail]", "Status": "Active"})

Table Class

For creating multi-sheet reports:

table = eywa.Table("MonthlyReport")
table.add_sheet(users_sheet)
table.add_sheet(stats_sheet)

# Convert to JSON for reporting
eywa.report("Monthly report", {"table": json.loads(table.toJSON())})

Constants

• SUCCESS - Task completed successfully
• ERROR - Task failed with error
• PROCESSING - Task is currently processing
• EXCEPTION - Task failed with exception

Complete Example

import asyncio
import eywa
import traceback

async def process_data():
    # Initialize
    eywa.open_pipe()
    
    try:
        # Get task info
        task = await eywa.get_task()
        eywa.info("Starting task", {"taskId": task["euuid"]})
        
        # Update status
        eywa.update_task(eywa.PROCESSING)
        
        # Query data with GraphQL
        result = await eywa.graphql("""
            query GetActiveUsers {
                searchUser(_where: {active: {_eq: true}}) {
                    euuid
                    name
                    email
                }
            }
        """)

        users = result["searchUser"]

        # Create report
        sheet = eywa.Sheet("ActiveUsers")
        sheet.set_columns(["ID", "Name", "Email"])
        
        for user in users:
            eywa.debug("Processing user", {"userId": user["euuid"]})
            sheet.add_row({
                "ID": user["euuid"],
                "Name": user["name"],
                "Email": user.get("email", "N/A")
            })
        
        # Report results
        eywa.report("Found active users", {
            "count": len(users),
            "sheet": sheet.__dict__
        })
        
        # Success!
        eywa.info("Task completed")
        eywa.close_task(eywa.SUCCESS)
        
    except Exception as e:
        eywa.error("Task failed", {
            "error": str(e),
            "traceback": traceback.format_exc()
        })
        eywa.close_task(eywa.ERROR)

if __name__ == "__main__":
    asyncio.run(process_data())

Type Hints

The library includes comprehensive type hints via .pyi file:

from typing import Dict, Any, Optional
import eywa

async def process() -> None:
    task: Dict[str, Any] = await eywa.get_task()
    result: Dict[str, Any] = await eywa.graphql(
        "query { searchUser { name } }", 
        variables={"limit": 10}
    )

Error Handling

The client includes custom exception handling:

try:
    result = await eywa.graphql("{ invalid }")
except eywa.JSONRPCException as e:
    eywa.error(f"GraphQL failed: {e.message}", {"error": e.data})

Testing

Test your robot locally using the EYWA CLI:

eywa run -c 'python my_robot.py'

Examples

To run examples, position terminal to root project folder and run:

# Test all features
python -m examples.test_eywa_client

# Run a simple GraphQL query
python -m examples.raw_graphql

# WebDriver example
python -m examples.webdriver

Requirements

• Python 3.7+
• Dependencies:
  • nanoid>=2.0.0 - For generating unique IDs
  • aiohttp>=3.8.0 - For async HTTP operations (file uploads/downloads)

License

MIT

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

Support

For issues and questions, please visit the EYWA repository.