1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
//! Client for a single Actor task (`/v2/actor-tasks/{actorTaskId}`).
use serde::Serialize;
use serde_json::Value;
use crate::client::ApifyClient;
use crate::clients::actor::ActorStartOptions;
use crate::clients::base::{
delete_resource, get_resource, post_with_body, update_resource, ResourceContext,
};
use crate::clients::run::{LastRunOptions, RunClient};
use crate::clients::run_collection::RunCollectionClient;
use crate::clients::webhook_collection::WebhookCollectionClient;
use crate::common::QueryParams;
use crate::error::ApifyClientResult;
use crate::http_client::HttpClient;
use crate::models::{ActorRun, Task};
/// Client for a specific Actor task.
#[derive(Debug, Clone)]
pub struct TaskClient {
root: ApifyClient,
ctx: ResourceContext,
}
impl TaskClient {
pub(crate) fn new(root: ApifyClient, http: HttpClient, base_url: &str, id: &str) -> Self {
Self {
root,
ctx: ResourceContext::single(http, base_url, "actor-tasks", id),
}
}
/// Fetches the task object, or `None` if it does not exist.
pub async fn get(&self) -> ApifyClientResult<Option<Task>> {
get_resource(&self.ctx, None, &QueryParams::new()).await
}
/// Updates the task with the given fields.
pub async fn update<T: Serialize>(&self, new_fields: &T) -> ApifyClientResult<Task> {
update_resource(&self.ctx, None, new_fields).await
}
/// Deletes the task.
pub async fn delete(&self) -> ApifyClientResult<()> {
delete_resource(&self.ctx, None).await
}
/// Starts the task and returns immediately with the created run.
///
/// `input` overrides the task's saved input (or `None` to use the saved input).
pub async fn start<T: Serialize>(
&self,
input: Option<&T>,
options: ActorStartOptions,
) -> ApifyClientResult<ActorRun> {
let mut params = QueryParams::new();
options.apply(&mut params);
let body = match input {
Some(value) => Some(serde_json::to_vec(value)?),
None => None,
};
post_with_body(&self.ctx, Some("runs"), ¶ms, body, "application/json").await
}
/// Starts the task and waits (client-side polling) for it to finish.
///
/// `wait_secs` controls the wait budget:
/// - `None` polls indefinitely until the run reaches a terminal state.
/// - `Some(n)` bounds the wait to roughly `n` seconds; if the run has not finished by
/// then, the **last fetched (still non-terminal) run is returned** rather than an
/// error. Check `status` / `is_terminal()` on the result when using `Some`.
pub async fn call<T: Serialize>(
&self,
input: Option<&T>,
options: ActorStartOptions,
wait_secs: Option<i64>,
) -> ApifyClientResult<ActorRun> {
let run = self.start(input, options).await?;
self.root.run(run.id).wait_for_finish(wait_secs).await
}
/// Fetches the task's saved input, or `None` if not set.
pub async fn get_input(&self) -> ApifyClientResult<Option<Value>> {
let response =
crate::clients::base::get_raw(&self.ctx, Some("input"), &QueryParams::new()).await?;
match response {
Some(r) => Ok(Some(serde_json::from_slice(&r.body)?)),
None => Ok(None),
}
}
/// Updates the task's saved input.
pub async fn update_input<T: Serialize>(&self, input: &T) -> ApifyClientResult<Value> {
let body = serde_json::to_vec(input)?;
let url = self.ctx.url(Some("input"));
let mut headers = std::collections::HashMap::new();
headers.insert("Content-Type".to_string(), "application/json".to_string());
let response = self
.ctx
.http
.call(crate::http_client::HttpRequest {
method: crate::http_client::HttpMethod::Put,
url,
headers,
body: Some(body),
timeout: crate::clients::base::DEFAULT_REQUEST_TIMEOUT,
})
.await?;
Ok(serde_json::from_slice(&response.body)?)
}
/// Returns a client for the last run of this task, optionally filtered by run status.
///
/// `status` filters by run status (e.g. `"SUCCEEDED"`, `"FAILED"`, `"RUNNING"`); pass `None`
/// to leave it unfiltered. This maps to the `status` query parameter on
/// `GET /v2/actor-tasks/{actorTaskId}/runs/last` and mirrors the reference client's
/// `lastRun({ status })`. To also filter by `origin`, use [`TaskClient::last_run_with_options`].
pub fn last_run(&self, status: Option<&str>) -> RunClient {
self.last_run_with_options(LastRunOptions {
status: status.map(str::to_owned),
origin: None,
})
}
/// Returns a client for the last run of this task, applying the given [`LastRunOptions`]
/// (e.g. [`LastRunOptions::status`] and/or [`LastRunOptions::origin`]).
///
/// `status` filters by run status (e.g. `"SUCCEEDED"`, `"FAILED"`, `"RUNNING"`) and is the
/// spec's documented filter on `GET /v2/actor-tasks/{actorTaskId}/runs/last`. `origin` filters
/// by how the run was started; accepted values are the platform's run origins (e.g.
/// `"DEVELOPMENT"`, `"WEB"`, `"API"`, `"SCHEDULER"`). `origin` is not declared by the OpenAPI
/// spec and is sent only for parity with the reference client's `lastRun({ status, origin })`.
/// Both are sent as query parameters; leave a field as `None` to omit it.
pub fn last_run_with_options(&self, options: LastRunOptions) -> RunClient {
let mut client = RunClient::new(
self.root.clone(),
self.ctx.http.clone(),
&self.ctx.url(None),
"runs",
"last",
);
if let Some(status) = options.status.as_deref() {
client.set_base_param("status", status);
}
if let Some(origin) = options.origin.as_deref() {
client.set_base_param("origin", origin);
}
client
}
/// Returns a client for this task's run collection.
pub fn runs(&self) -> RunCollectionClient {
RunCollectionClient::new(self.ctx.http.clone(), &self.ctx.url(None), "runs")
}
/// Returns a client for this task's webhook collection.
pub fn webhooks(&self) -> WebhookCollectionClient {
WebhookCollectionClient::with_base(self.ctx.http.clone(), &self.ctx.url(None))
}
}