Grabba Python SDK

Grabba Python SDK provides a simple and intuitive interface for scheduling web data extraction jobs, estimating costs, retrieving job results, and managing your workflows. All SDK types are implemented as Pydantic BaseModels for full JSON compatibility and built-in validation.

---

Installation

Install the SDK using pip:

pip install grabba

---

Basic Setup

Import the Client and Required Types

from grabba import (
    Grabba, Job, JobNavigationType, JobSchedulePolicy, JobTaskType,
    KnowledgeBase, ContextResult
)

Note: All types such as Job, JobTask, etc., are now Pydantic BaseModels. They support .model_dump() and .json() for serialization, and Enum fields are automatically converted to literal values.

Initialize a Client Instance

grabba = Grabba(api_key="your-api-key", region="US")  # Optional: Defaults to US

---

Methods

extract

grabba.extract(job: Job) -> JobExecutionResponse

Schedules a new web data extraction job for immediate execution.

---

schedule_job

grabba.schedule_job(job_id: str) -> JobExecutionResponse

Schedules an already-created job for execution.

---

wait_for_job_completion

grabba.wait_for_job_completion(job_id: str, timeout_seconds: int = 240, poll_interval_seconds: int = 5) -> JobWaitResponse

Waits for a long-running job to reach a terminal state (completed, failed, cancelled) using SSE first with polling fallback.

Returns a structured payload including:

• job_id
• status
• completed
• timed_out
• source (sse, fetch_specific_job, timeout)
• reason
• optional job_result_id

Example:

wait = grabba.wait_for_job_completion(job_id, timeout_seconds=300, poll_interval_seconds=5)
if wait.completed and wait.job_result_id:
    result = grabba.get_job_result(wait.job_result_id)

---

get_job_health

grabba.get_job_health(job_id: str) -> JobHealthResponse

Fetches a safe health snapshot for a job and recommended safe actions.

Example:

health = grabba.get_job_health(job_id)
print(health.health.safe_actions)

---

retry_job_safe

grabba.retry_job_safe(job_id: str) -> BaseResponse

Safely schedules an immediate retry for failed/stalled jobs.

---

resume_schedule_safe

grabba.resume_schedule_safe(job_id: str) -> GetJobResponse

Safely resumes a paused recurring schedule.

---

pause_schedule_with_reason

grabba.pause_schedule_with_reason(job_id: str, note: str | None = None) -> GetJobResponse

Pauses a schedule through guarded controls with an optional human-readable reason.

---

get_jobs

grabba.get_jobs(page=1, limit=25) -> GetJobsResponse

Fetches a paginated list of your submitted jobs.

---

get_job

grabba.get_job(job_id: str) -> GetJobResponse

Fetches details of a specific job.

---

get_job_result

grabba.get_job_result(job_result_id: str) -> GetJobResultResponse

Retrieves the result of a specific completed job.

---

delete_job

grabba.delete_job(job_id: str) -> BaseResponse

Deletes a job using its job ID.

---

delete_job_result

grabba.delete_job_result(job_result_id: str) -> BaseResponse

Deletes a specific job result.

---

create_job

grabba.create_job(job: Job) -> JobCreationResponse

Creates a job but does not execute it immediately.

---

estimate_job_cost

grabba.estimate_job_cost(job: Job) -> JobEstimatedCostResponse

Estimates the cost of running a job based on its tasks, duration, and region.

---

get_stats

grabba.get_stats() -> JobStatsResponse

Fetches job usage stats for your API key.

---

get_available_regions

grabba.get_available_regions() -> List[Dict[str, str]]

Returns a list of available regions supported by the scraping engine.

---

Knowledge Base Methods

create_knowledge_base

grabba.create_knowledge_base(name: str, description: str = None) -> CreateKnowledgeBaseResponse

Creates a new knowledge base.

---

get_knowledge_bases

grabba.get_knowledge_bases() -> GetKnowledgeBasesResponse

Retrieves a list of all your knowledge bases.

---

get_knowledge_base

grabba.get_knowledge_base(kb_id: str) -> GetKnowledgeBaseResponse

Retrieves a specific knowledge base by its ID.

---

delete_knowledge_base

grabba.delete_knowledge_base(kb_id: str) -> BaseResponse

Deletes a specific knowledge base by its ID.

---

store_context

grabba.store_context(kb_id: str, context: str, metadata: Dict = None) -> StoreContextResponse

Stores a new context (piece of information) into a knowledge base.

---

fetch_context

grabba.fetch_context(kb_id: str, query: str, options: Dict = None) -> FetchContextResponse

Fetches relevant context from a knowledge base based on a semantic query.

---

update_context

grabba.update_context(context_id: str, content: str = None, metadata: Dict = None) -> UpdateContextResponse

Updates a specific context entry by its ID.

---

delete_context

grabba.delete_context(context_id: str) -> BaseResponse

Deletes a specific context entry by its ID.

---

gather_context

grabba.gather_context(kb_id: str) -> GatherContextResponse

Gathers and returns all unique metadata tags and values for a given knowledge base.

---

Error Handling

The SDK throws exceptions for:

• Invalid API keys
• BadRequest responses (400)
• Network issues
• Validation errors in input models

Example:

try:
    response = grabba.extract(job)
    print(response.job_result)
except Exception as e:
    print("Failed to run job:", e)

---

Contributing

Contributions are welcome! Please open an issue or submit a pull request on GitHub.

---

License

MIT License. See the LICENSE file.

---

Additional Notes

• Pydantic Serialization: All SDK models support .json() or .model_dump() for easy JSON conversion.

• Enum Fields: Enum values like JobSchedulePolicy.IMMEDIATELY are serialized automatically to their string literals (e.g., "IMMEDIATELY").

• Type Safety: All inputs are strictly validated using Pydantic.

---

Change Log

Version 0.0.15 (Latest)

• New Method - get_job_health(job_id: str): Return guarded health metadata and safe actions.
• New Method - retry_job_safe(job_id: str): Trigger safe retry for failed/stalled jobs.
• New Method - resume_schedule_safe(job_id: str): Resume schedules through guarded endpoint.
• New Method - pause_schedule_with_reason(job_id: str, note: str | None): Pause schedules with explicit note through guarded endpoint.
• New Method - wait_for_job_completion(...): SSE-first completion wait with polling fallback.
• Webhook + scheduling controls stabilized across SDK surface.

Version 0.0.14 and below

• Core extraction, scheduling, knowledge base, and webhook features.
• Pydantic model exports and response normalization improvements.

---

Let me know if you'd like a Markdown export or to update the repo structure with examples/docs as well.