Error handling idioms

from pydantic import BaseModel
from imperal_sdk import ActionResult
from imperal_sdk.errors import NotFoundError


class GetNoteParams(BaseModel):
    note_id: str


async def fn_get_note(ctx, params: GetNoteParams) -> ActionResult:
    # Graceful path — return ActionResult.error for expected failures
    if not params.note_id:
        return ActionResult.error("note_id is required")

    doc = await ctx.store.get("notes", params.note_id)
    if doc is None:
        return ActionResult.error(f"Note '{params.note_id}' not found")

    return ActionResult.success(
        data={"note_id": doc.id, "title": doc.data.get("title", "")},
        summary="Note retrieved",
    )

from imperal_sdk import ActionResult

# Factory signature (source: imperal_sdk.types.action_result)
result = ActionResult.error("Something went wrong", retryable=False)

from pydantic import BaseModel
from imperal_sdk import ActionResult


class FetchInboxParams(BaseModel):
    limit: int = 25


async def fn_fetch_inbox(ctx, params: FetchInboxParams) -> ActionResult:
    try:
        resp = await ctx.http.get(
            "https://mail-api.example.com/inbox",
            params={"limit": params.limit},
        )
    except Exception:
        # Transient — network timeout, DNS failure, etc.
        return ActionResult.error(
            "Could not reach the mail service. Please try again.",
            retryable=True,
        )

    if not resp.ok:
        if resp.status_code == 429:
            # Transient — rate limited
            return ActionResult.error(
                "Mail API rate limit reached. Please try again in a moment.",
                retryable=True,
            )
        # Permanent — bad credentials, server error, etc.
        return ActionResult.error(
            f"Mail API returned an error ({resp.status_code})."
        )

    return ActionResult.success(
        data={"messages": resp.json().get("messages", [])},
        summary="Inbox loaded",
    )

from pydantic import BaseModel
from imperal_sdk import ActionResult


class CreateNoteParams(BaseModel):
    title: str


async def fn_create_note_bad(ctx, params: CreateNoteParams) -> ActionResult:
    try:
        result = await ctx.http.post("/notes", json={"title": params.title})
    except Exception as e:
        # ❌ str(e) may contain internal paths, class names, or PII
        return ActionResult.error(str(e))
    return ActionResult.success(data={}, summary="Done")

from pydantic import BaseModel
from imperal_sdk import ActionResult
import logging

log = logging.getLogger(__name__)


class CreateNoteParams(BaseModel):
    title: str


async def fn_create_note_good(ctx, params: CreateNoteParams) -> ActionResult:
    try:
        resp = await ctx.http.post("/notes", json={"title": params.title})
    except Exception as exc:
        # ✅ Log the raw detail internally; show a stable message to the user
        log.error("create_note http failed: %s", exc)
        return ActionResult.error(
            "Could not create the note. Please try again.",
            retryable=True,
        )
    if not resp.ok:
        log.warning("create_note backend error %s", resp.status_code)
        return ActionResult.error("Note creation failed. Please check your input.")
    note_id = resp.json().get("id", "")
    return ActionResult.success(
        data={"note_id": note_id, "title": params.title},
        summary=f"Note '{params.title}' created",
        refresh_panels=["sidebar"],
    )

from imperal_sdk.errors import ImperalError

from imperal_sdk.errors import SkeletonAccessForbidden

from imperal_sdk.chat import TaskCancelled

from pydantic import BaseModel
from imperal_sdk import ActionResult
from imperal_sdk.chat import TaskCancelled
import logging

log = logging.getLogger(__name__)


class ExportParams(BaseModel):
    format: str = "csv"


async def fn_export_data(ctx, params: ExportParams) -> ActionResult:
    try:
        rows = await ctx.store.query("records")
        await ctx.progress(25, "Fetched records…")

        processed = []
        for i, row in enumerate(rows.data):
            processed.append(row.data)
            if i % 50 == 0:
                await ctx.progress(25 + int(50 * i / max(len(rows.data), 1)), "Processing…")

        await ctx.progress(90, "Formatting output…")
        # ... format and upload

    except TaskCancelled:
        # ✅ User cancelled — return a clean graceful message
        return ActionResult.error("Export cancelled")

    except Exception as exc:
        log.error("export failed: %s", exc)
        return ActionResult.error(
            "Export failed. Please try again.",
            retryable=True,
        )

    await ctx.progress(100, "Done")
    return ActionResult.success(
        data={"row_count": len(processed)},
        summary=f"Exported {len(processed)} records",
    )

from pydantic import BaseModel
from imperal_sdk import ActionResult
from imperal_sdk.errors import RateLimitError, APIError
import logging

log = logging.getLogger(__name__)


class CreateTaskParams(BaseModel):
    title: str
    project_id: int


async def fn_create_task(ctx, params: CreateTaskParams) -> ActionResult:
    try:
        resp = await ctx.http.post(
            "https://tasks.internal/tasks",
            json={"title": params.title, "project_id": params.project_id},
        )
    except RateLimitError as e:
        return ActionResult.error(
            f"Task service is rate-limited. Retry in {e.retry_after}s.",
            retryable=True,
        )
    except Exception as exc:
        log.error("create_task failed: %s", exc)
        return ActionResult.error(
            "Could not reach the task service. Please try again.",
            retryable=True,
        )

    if not resp.ok:
        return ActionResult.error(
            _translate_task_error(resp.json().get("detail", "Task creation failed"))
        )

    task_id = resp.json().get("id")
    return ActionResult.success(
        data={"task_id": task_id, "title": params.title},
        summary=f"Task '{params.title}' created",
        refresh_panels=["sidebar"],
    )


def _translate_task_error(detail: str) -> str:
    """Translate backend error details into user-friendly messages."""
    if "project not found" in detail.lower():
        return "Project not found. Please select a valid project."
    if "duplicate" in detail.lower():
        return "A task with this title already exists in the project."
    # Fall back to a generic message — do not surface raw backend detail
    return "Task creation failed. Please check your input."

from pydantic import BaseModel, Field
from imperal_sdk import ActionResult


class UpdateNoteParams(BaseModel):
    note_id: str
    title: str = Field(default="")
    content: str = Field(default="")


async def fn_update_note(ctx, params: UpdateNoteParams) -> ActionResult:
    # Guard clauses — fast exits before any I/O
    if not params.note_id:
        return ActionResult.error("note_id is required")
    if not params.title and not params.content:
        return ActionResult.error("At least one of title or content is required")

    # Only reach I/O after guards pass
    try:
        resp = await ctx.http.patch(
            f"/notes/{params.note_id}",
            json={"title": params.title, "content": params.content},
        )
    except Exception:
        return ActionResult.error("Note update failed. Please try again.", retryable=True)

    if not resp.ok:
        if resp.status_code == 404:
            return ActionResult.error(f"Note '{params.note_id}' not found")
        return ActionResult.error("Note update failed")

    return ActionResult.success(
        data={"note_id": params.note_id},
        summary="Note updated",
        refresh_panels=["sidebar", "editor"],
    )

from pydantic import BaseModel
from imperal_sdk import ActionResult
import base64
import logging

log = logging.getLogger(__name__)


class UploadAttachmentParams(BaseModel):
    note_id: str
    filename: str
    data_base64: str


async def fn_upload_attachment(ctx, params: UploadAttachmentParams) -> ActionResult:
    if not params.note_id:
        return ActionResult.error("note_id is required")
    if not params.data_base64:
        return ActionResult.error("No file data received")

    # Step 1: decode — non-retryable, user input error
    try:
        raw_bytes = base64.b64decode(params.data_base64)
    except Exception:
        return ActionResult.error("Invalid file data (base64 decode failed)")

    # Step 2: upload — potentially transient
    try:
        file_info = await ctx.storage.upload(
            f"attachments/{params.note_id}/{params.filename}",
            raw_bytes,
        )
    except Exception as exc:
        log.error("attachment upload failed: %s", exc)
        return ActionResult.error(
            "File upload failed. Please try again.",
            retryable=True,
        )

    # Step 3: link to note — could fail without affecting the upload
    try:
        await ctx.store.update("notes", params.note_id, {
            "attachments": [{"path": file_info.path, "name": params.filename}],
        })
    except Exception as exc:
        log.warning("attachment link failed (file uploaded): %s", exc)
        # File is uploaded but not linked — surface this to the user
        return ActionResult.error(
            "File uploaded but could not be linked to the note. "
            "Please refresh and try attaching again."
        )

    return ActionResult.success(
        data={"path": file_info.path, "filename": params.filename},
        summary=f"Attached '{params.filename}'",
        refresh_panels=["editor"],
    )

from pydantic import BaseModel
from imperal_sdk import ActionResult


class RunQueryParams(BaseModel):
    query: str


async def fn_run_query(ctx, params: RunQueryParams) -> ActionResult:
    await ctx.log(f"run_query called: user={ctx.user.imperal_id}", level="info")

    try:
        result = await ctx.http.post("/query", json={"q": params.query})
    except Exception as exc:
        await ctx.log(f"run_query failed: {exc}", level="error")
        return ActionResult.error("Query failed. Please try again.", retryable=True)

    return ActionResult.success(
        data={"rows": result.json().get("rows", [])},
        summary="Query completed",
    )

from pydantic import BaseModel, Field, model_validator
from imperal_sdk import ActionResult
from typing import Optional


class CreateNoteParams(BaseModel):
    title: str = Field(..., min_length=1, max_length=500)
    folder_id: Optional[str] = None
    is_pinned: bool = False

    @model_validator(mode="after")
    def title_not_whitespace(self) -> "CreateNoteParams":
        if not self.title.strip():
            raise ValueError("title must not be blank")
        return self


async def fn_create_note(ctx, params: CreateNoteParams) -> ActionResult:
    # By here, Pydantic has already enforced title length and non-blank.
    # Only handle runtime / database-state validation manually.
    if params.folder_id:
        folder = await ctx.store.get("folders", params.folder_id)
        if folder is None:
            return ActionResult.error(
                f"Folder '{params.folder_id}' not found. "
                "List your folders to find the correct ID."
            )

    doc = await ctx.store.create("notes", {
        "title": params.title.strip(),
        "folder_id": params.folder_id,
        "is_pinned": params.is_pinned,
    })

    return ActionResult.success(
        data={"note_id": doc.id, "title": doc.data["title"]},
        summary=f"Created note '{doc.data['title']}'",
        refresh_panels=["sidebar"],
    )

from pydantic import BaseModel
from imperal_sdk import ActionResult
from imperal_sdk.types import HTTPResponse
import logging

log = logging.getLogger(__name__)


class ListCampaignsParams(BaseModel):
    account_id: str


async def fn_list_campaigns(ctx, params: ListCampaignsParams) -> ActionResult:
    try:
        resp: HTTPResponse = await ctx.http.get(
            f"/accounts/{params.account_id}/campaigns",
        )
    except Exception as exc:
        log.error("list_campaigns transport error: %s", exc)
        return ActionResult.error(
            "Could not reach the campaigns service. Please try again.",
            retryable=True,
        )

    if resp.status_code == 404:
        return ActionResult.error(f"Account '{params.account_id}' not found")
    if resp.status_code == 401:
        return ActionResult.error(
            "Authentication failed. Please reconnect your account in settings."
        )
    if resp.status_code == 429:
        return ActionResult.error(
            "API rate limit reached. Please try again in a few minutes.",
            retryable=True,
        )
    if not resp.ok:
        log.warning("list_campaigns non-OK: %s", resp.status_code)
        return ActionResult.error(
            f"Campaigns API returned an error ({resp.status_code}). "
            "Please try again.",
            retryable=resp.status_code >= 500,
        )

    campaigns = resp.json().get("campaigns", [])
    return ActionResult.success(
        data={"campaigns": campaigns, "total": len(campaigns)},
        summary=f"Found {len(campaigns)} campaigns",
    )

from imperal_sdk.types import HTTPResponse
from imperal_sdk import ActionResult
import logging

log = logging.getLogger(__name__)


async def api_call(ctx, method: str, path: str, **kwargs) -> HTTPResponse:
    """Central HTTP wrapper. Raises on transport error; returns HTTPResponse otherwise."""
    try:
        fn = getattr(ctx.http, method)
        return await fn(path, **kwargs)
    except Exception as exc:
        log.error("api_call %s %s failed: %s", method.upper(), path, exc)
        raise


def is_no_connection_error(resp: dict) -> bool:
    """Return True when the bridge indicates the user has no account connected."""
    detail = resp.get("detail", "")
    return isinstance(detail, str) and "connect" in detail.lower()

from pydantic import BaseModel, Field
from imperal_sdk import ActionResult
from imperal_sdk.errors import APIError
import logging

log = logging.getLogger(__name__)


class SaveSettingsParams(BaseModel):
    theme: str = Field(default="system")
    notifications_enabled: bool = True


async def fn_save_settings(ctx, params: SaveSettingsParams) -> ActionResult:
    settings_data = {
        "theme": params.theme,
        "notifications_enabled": params.notifications_enabled,
    }

    try:
        existing = await ctx.store.get("settings", "user_prefs")
        if existing:
            await ctx.store.update("settings", "user_prefs", settings_data)
        else:
            await ctx.store.create("settings", settings_data)
    except APIError as exc:
        log.error("save_settings store error %s: %s", exc.status_code, exc.message)
        return ActionResult.error(
            "Settings could not be saved. Please try again.",
            retryable=True,
        )
    except Exception as exc:
        log.error("save_settings unexpected: %s", exc)
        return ActionResult.error("Settings could not be saved.")

    return ActionResult.success(
        data=settings_data,
        summary="Settings saved",
        refresh_panels=["settings"],
    )

from pydantic import BaseModel
from imperal_sdk import ActionResult
import logging

log = logging.getLogger(__name__)


class InboxSummary(BaseModel):
    """Registered via @ext.cache_model("inbox_summary") in app.py."""
    items: list = []


class ListItemsParams(BaseModel):
    force_refresh: bool = False


async def fn_list_items(ctx, params: ListItemsParams) -> ActionResult:
    # Cache miss is expected and normal — not an error
    cached = None
    if not params.force_refresh:
        try:
            cached = await ctx.cache.get("inbox_summary", InboxSummary)
        except Exception as exc:
            # Cache transport error — degrade gracefully, fetch live
            log.warning("cache get failed, fetching live: %s", exc)

    if cached is None:
        try:
            resp = await ctx.http.get("/inbox")
        except Exception as exc:
            log.error("inbox fetch failed: %s", exc)
            return ActionResult.error(
                "Could not load inbox. Please try again.",
                retryable=True,
            )
        items = resp.json().get("items", [])
    else:
        items = cached.items

    return ActionResult.success(
        data={"items": items},
        summary=f"Loaded {len(items)} items",
    )

from pydantic import BaseModel
from imperal_sdk import ActionResult, CircularCallError
from imperal_sdk.errors import NotFoundError, AuthError, ExtensionError
import logging

log = logging.getLogger(__name__)


class SendWithAttachmentParams(BaseModel):
    note_id: str
    recipient: str


async def fn_send_note_as_email(ctx, params: SendWithAttachmentParams) -> ActionResult:
    # Fetch note content via cross-extension call
    try:
        note = await ctx.extensions.call("notes", "get_note_content", note_id=params.note_id)
    except NotFoundError:
        return ActionResult.error(
            f"Note '{params.note_id}' not found. Cannot send."
        )
    except AuthError:
        return ActionResult.error(
            "This extension does not have permission to read notes."
        )
    except CircularCallError as exc:
        log.error("circular extension call: %s", exc)
        return ActionResult.error("Internal configuration error. Please contact support.")
    except ExtensionError as exc:
        log.error("extensions.call failed: %s", exc)
        return ActionResult.error("Could not load note content. Please try again.", retryable=True)

    # Continue with send logic...
    return ActionResult.success(
        data={"sent": True},
        summary=f"Note sent to {params.recipient}",
    )

from pydantic import BaseModel
from imperal_sdk import Extension, ActionResult
import hmac, hashlib, logging

ext = Extension(
    "my-ext",
    display_name="My Extension",
    description="Example webhook handler with HMAC verification.",
    actions_explicit=True,
)

log = logging.getLogger(__name__)
WEBHOOK_SECRET = "loaded-from-env"


@ext.webhook("/events", secret_header="X-Webhook-Signature")
async def on_webhook(ctx, headers: dict, body: str, query_params: dict) -> ActionResult:
    # Verify HMAC signature — your responsibility; SDK provides no helper
    sig = headers.get("x-webhook-signature", "")
    expected = "sha256=" + hmac.new(
        WEBHOOK_SECRET.encode(), body.encode(), hashlib.sha256
    ).hexdigest()
    if not hmac.compare_digest(sig, expected):
        # Return ActionResult.error for invalid signature
        log.warning("webhook: invalid signature from %s", headers.get("host"))
        return ActionResult.error("Invalid webhook signature")

    try:
        payload = __import__("json").loads(body)
    except Exception:
        return ActionResult.error("Invalid JSON payload")

    event_type = payload.get("type", "unknown")
    log.info("webhook received event: %s", event_type)

    # Process event...
    return ActionResult.success(data={"event_type": event_type}, summary="Webhook processed")

from imperal_sdk import Extension, ActionResult
import logging

ext = Extension(
    "my-ext",
    display_name="My Extension",
    description="Example scheduled job with per-user isolation.",
    actions_explicit=True,
)

log = logging.getLogger(__name__)


@ext.schedule("daily_digest", cron="0 9 * * *")
async def run_daily_digest(ctx) -> None:
    """Send daily digest to all active users."""
    errors = []

    async for user_id in ctx.store.list_users("preferences"):
        user_ctx = ctx.as_user(user_id)
        try:
            await _send_digest_for_user(user_ctx)
        except Exception as exc:
            # Isolate — log the failure but continue to next user
            log.error("digest failed for user %s: %s", user_id, exc)
            errors.append(user_id)

    if errors:
        log.warning("digest completed with %d failures: %s", len(errors), errors[:5])
    else:
        log.info("digest completed successfully")


async def _send_digest_for_user(ctx) -> None:
    prefs = await ctx.store.get("preferences", "digest")
    if prefs is None or not prefs.data.get("enabled"):
        return
    # ... fetch and send
    await ctx.notify("Your daily digest is ready", priority="low")

from imperal_sdk import Extension
import logging

ext = Extension(
    "my-ext",
    display_name="My Extension",
    description="Example event handler with graceful degradation.",
    actions_explicit=True,
)

log = logging.getLogger(__name__)


@ext.on_event("notes.created")
async def on_note_created(ctx, event) -> None:
    """React to a note creation event.

    Failures here are silent to the user — log everything.
    No retry will be issued.
    """
    try:
        note_id = event.data.get("note_id")
        if not note_id:
            log.warning("on_note_created: event missing note_id, skipping")
            return

        # Attempt secondary action — cache invalidation, indexing, etc.
        await ctx.store.update("indexes", "notes_count", {"invalidated": True})
    except Exception as exc:
        # Must not raise — that would silently drop the event
        log.error("on_note_created handler error: %s", exc)

from imperal_sdk import ActionResult
import logging

log = logging.getLogger(__name__)


async def _bad_example(ctx, e: Exception) -> ActionResult:
    # ❌ Wrong — exposes internal exception details
    return ActionResult.error(str(e))


async def _good_example(ctx, e: Exception) -> ActionResult:
    # ✅ Correct — stable user-facing message
    log.error("operation failed: %s", e)
    return ActionResult.error("Operation failed. Please try again.", retryable=True)

from imperal_sdk import ActionResult
import logging

log = logging.getLogger(__name__)


async def _bad_update(ctx, item_id: str, data: dict) -> None:
    # ❌ Wrong — failure is invisible
    try:
        await ctx.store.update("items", item_id, data)
    except Exception:
        pass  # silently ignored


async def _good_update(ctx, item_id: str, data: dict) -> ActionResult:
    # ✅ Correct — log and surface
    try:
        await ctx.store.update("items", item_id, data)
    except Exception as exc:
        log.error("update failed for %s: %s", item_id, exc)
        return ActionResult.error("Update failed. Please try again.", retryable=True)
    return ActionResult.success(data={}, summary="Updated")

from imperal_sdk import ActionResult


async def _bad_bare_except(ctx) -> ActionResult:
    # ❌ Wrong — catches BaseException including KeyboardInterrupt, SystemExit
    try:
        pass  # your operation here
    except:  # noqa: E722
        return ActionResult.error("Failed")
    return ActionResult.success(data={}, summary="Done")


async def _good_except(ctx) -> ActionResult:
    # ✅ Correct — catch Exception only
    try:
        pass  # your operation here
    except Exception:
        return ActionResult.error("Failed")
    return ActionResult.success(data={}, summary="Done")

from imperal_sdk import Extension
import logging

ext = Extension(
    "mail-ext",
    display_name="Mail",
    description="Example showing correct event handler error handling.",
    actions_explicit=True,
)
log = logging.getLogger(__name__)


async def _process_mail_stub(event: object) -> bool:
    return True  # placeholder


# ❌ Wrong — exception is swallowed, produces no feedback
@ext.on_event("mail.received.bad")
async def on_mail_bad(ctx, event) -> None:  # type: ignore[no-untyped-def]
    result = await _process_mail_stub(event)
    if not result:
        raise RuntimeError("processing failed")  # silently lost


# ✅ Correct — log and return
@ext.on_event("mail.received")
async def on_mail(ctx, event) -> None:  # type: ignore[no-untyped-def]
    try:
        result = await _process_mail_stub(event)
        if not result:
            log.warning("on_mail: processing returned empty result")
    except Exception as exc:
        log.error("on_mail handler error: %s", exc)

from pydantic import BaseModel
from imperal_sdk import Extension, ChatExtension, ActionResult

ext = Extension(
    "notes-ext",
    display_name="Notes",
    description="Example showing correct skeleton access patterns.",
    actions_explicit=True,
)
chat = ChatExtension(
    ext,
    tool_name="tool_notes_chat",
    description="AI chat interface for the notes extension",
)


class SearchParams(BaseModel):
    query: str


# ❌ Wrong — raises SkeletonAccessForbidden at runtime (do not run this)
# @chat.function("search_notes_bad", description="Search notes — wrong pattern",
#                action_type="read")
# async def fn_search_bad(ctx, params: SearchParams) -> ActionResult:
#     schema = await ctx.skeleton.get("notes")  # SkeletonAccessForbidden!
#     ...


# ✅ Correct — read live data from ctx.store or ctx.http
@chat.function(
    "search_notes",
    description="Search notes by title using live store data",
    action_type="read",
)
async def fn_search(ctx, params: SearchParams) -> ActionResult:
    # Use ctx.store for live data; skeleton is for the LLM classifier only
    results = await ctx.store.query("notes", where={"title": params.query})
    return ActionResult.success(
        data={"results": [d.data for d in results.data]},
        summary=f"Found {len(results.data)} notes",
    )

from imperal_sdk import Extension, ActionResult
import logging

ext = Extension(
    "notes-ext",
    display_name="Notes",
    description="Example showing correct skeleton failure handling.",
    actions_explicit=True,
)
log = logging.getLogger(__name__)


# ❌ Wrong — skeleton returning ActionResult.error shape
@ext.skeleton("notes_bad", ttl=300)
async def skeleton_refresh_notes_bad(ctx) -> dict:
    try:
        pass  # fetch data
    except Exception:
        return ActionResult.error("failed").to_dict()  # wrong shape!
    return {"response": {}}


# ✅ Correct — return safe zero values
@ext.skeleton("notes", ttl=300)
async def skeleton_refresh_notes(ctx) -> dict:
    count = 0
    try:
        count = await ctx.store.count("notes")
        return {"response": {"total": count}}
    except Exception as exc:
        log.error("skeleton refresh failed: %s", exc)
        return {"response": {"total": 0, "recent": []}}