Files
windmill/rust-client/src/client.rs
T
Diego Imbert 4427a3d37f feat: add workspace-specific flag for resources and variables (#8836)
* feat: add workspace-specific flag for resources and variables

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix: remove set_ws_specific endpoint and fix rust-client compilation

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix: fall back to workspace name for ws_specific file naming

When wsNameForFiles is not set (no wmill.yaml workspace config),
ws_specific items would not get workspace-suffixed filenames during
pull. Now falls back to workspace.name/workspaceId.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix: use workspace ID instead of CLI name for ws_specific file naming

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix: pass workspace ID fallback to elementsToMap for ws_specific push

Without this, workspace-specific files (e.g., a.admins.resource.yaml)
were not recognized during push when no wmill.yaml or git branch was
available, causing spurious deletions.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* ui nits

* nit

* Fix variable edit when only editing ws_specific

* mark_linked_variables_ws_specific

* Helper label

* Support json format alongside yaml

* Fix file naming push/pull asymetry & ws_specific orphans

* Revert all CLI diffs

* CLI now appends the remote ws_specific list to the local specificItems

* UI for Env switcher

* Refactor Resource/Variable editors to use dumb component

* Refactor side effects

* Editor works with multi workspaces

* Fix can_save

* Fix As JSON

* nit

* UI nits

* list_ws_specific_versions as pl sql function to avoid round trips

* UI Nits

* Per-workspace version read-only check

* fix: reset session context in list_ws_specific_versions to prevent RLS leakage

The function calls set_session_context() in a loop. Although SET LOCAL is
transaction-scoped (so settings revert at autocommit), defending against
the function being invoked inside a longer outer transaction:

- wrap the loop in a sub-block with EXCEPTION WHEN OTHERS that resets
  the session to a deny-default (windmill_user, empty session.* GUCs)
  before re-raising,
- on the happy path, reset to the same deny-default at the end of the
  function.

* feat: audit auto-marked ws_specific variables

When a resource is saved as ws_specific, every variable referenced via
$var: inside its value is auto-INSERTed into ws_specific. Previously
this happened silently. Now:

- mark_linked_variables_ws_specific takes the authed user,
- the INSERT uses RETURNING path so we know exactly which variables
  were freshly flipped (not the ones already ws_specific),
- each newly flipped variable gets a 'variables.set_ws_specific' audit
  entry pointing at the resource that triggered it.

* perf: skip mark_linked_variables_ws_specific when nothing relevant changed

update_resource was calling mark_linked_variables_ws_specific on every
save when the resource was ws_specific, even on a description-only or
label-only edit. Gate the call on `ns.value.is_some() || ns.ws_specific
== Some(true)` so we only re-mark when the $var: refs could actually
have changed or ws_specific was freshly enabled.

* docs: explain asymmetric ws_specific toggle in resource tooltip

Enabling the resource's 'Workspace specific' toggle silently marks
every variable referenced via $var: inside the value as ws_specific,
but disabling it does not un-mark those variables (they may be
referenced by other resources). Surface this in the tooltip so users
know what to expect.

* fix: surface non-404 errors when fetching ws_specific items in CLI sync

mergeWsSpecificFromServer was catching every error from listWsSpecific
and logging it at debug. That's correct for old servers without the
endpoint (404), but a 401/403/network failure would silently produce an
incomplete sync. Now distinguish 404 (debug, expected) from everything
else (warn with status + message) so users notice when the merge fails
for real reasons.

* perf: collapse compare_two_variables presence checks into one round-trip

The early-return path was issuing four sequential EXISTS queries
(ws_specific × {source, fork}, variable × {source, fork}). Combine
them into a single SELECT so the per-variable diff cost drops ~4x.

* sqlx prepare

* docs: clarify has_sql_updates invariant in update_variable

The else branch of the npath resolution is only reachable for non-rename
edits (labels-only, ws_specific-only) because ns.path being Some always
forces has_sql_updates=true at the top of the function. Add a debug_assert
and a comment explaining the invariant so a future change that decouples
ns.path from has_sql_updates trips immediately. Also use `path` directly
instead of unwrap_or_default-ing ns.path, since we know it's None here.

* chore: drop redundant ws_specific type augmentations

ListableResource and ListableVariable from $lib/gen now include
`ws_specific?: boolean` after the openapi.yaml additions in this
branch were regenerated. The intersection types in resources/+page
and variables/+page were duplicating the field — drop them.

* Put WsSpecificVersions toggle in top drawer bar

* nit size

* feat: detect local-only ws_specific items on sync push

When wmill.yaml lists a resource/variable in specificItems but the
remote isn't yet marked ws_specific for that item, sync push silently
dropped the flag because:

1. file-content diff alone never noticed (ws_specific is metadata, not
   YAML body) — push{Resource,Variable} were never called for those
   items;
2. even when called, isSuperset(local, remote) returned true and the
   early-return skipped the API call.

Now:

- mergeWsSpecificFromServer returns the raw server list alongside the
  merged config so push can compare 'in local' vs 'in server';
- a new computeWsSpecificFlagOnlyPushes helper walks the local file map,
  finds ws_specific-flagged paths absent from the server list, and the
  push function injects them as synthetic 'edited' changes (same before
  and after content) so the standard display + apply pipeline picks
  them up;
- push{Resource,Variable} no longer early-return when content matches
  but the ws_specific flag differs.

Pull is unaffected — only the push-side caller of mergeWsSpecific takes
the new (merged, serverItems) tuple.

* getDeployTo for selected ws

* refactor: ws_specific kind handling, support .json files

The ws_specific helpers had two warts:

1. computeWsSpecificFlagOnlyPushes hardcoded `.resource.{yaml,json}` /
   `.variable.{yaml,json}` magic strings, even though the existing
   getTypeStrFromPath / removeType helpers already do that work and
   already cover both extensions.
2. isSpecificItem / isItemTypeConfigured only matched `.yaml` paths,
   so users with opts.json local files got no specificItems coverage
   at all — patterns from wmill.yaml (and from mergeWsSpecificFromServer)
   are expressed with `.yaml`, and a `.json` file never matched.

Changes:

- Replace WS_SPECIFIC_KIND_MAP (a closed enum of resource+variable)
  with configKeyForItemKind, a generic kind→SpecificItemsConfig key
  mapping. Triggers fold into 'triggers' via the `_trigger` suffix,
  so adding a kind to the backend's list_ws_specific_versions doesn't
  require a CLI change.
- mergeWsSpecificFromServer now appends `${item.path}.${item.item_kind}.yaml`
  through the same helper.
- computeWsSpecificFlagOnlyPushes uses getTypeStrFromPath + removeType,
  gated by configKeyForItemKind. No more magic strings.
- isSpecificItem and isItemTypeConfigured normalize trailing `.json` to
  `.yaml` once at the entry, so a single set of patterns covers both
  extensions for the same logical item.

* refactor: dedicated change type for ws_specific flag-only pushes

Previously the sync push code injected a synthetic 'edited' Change with
before === after to nudge the apply loop into calling pushResource /
pushVariable for ws_specific-flag-only diffs, and a guard inside those
two functions skipped the early-return when the flag differed. The
contract was implicit and easy to break — any future 'skip identical
edits' optimization in the change pipeline would silently drop these
pushes.

Replace with an explicit Change variant:

  type WsSpecificFlag = {
    name: 'ws_specific_flag';
    path: string;
    kind: string;
    wsSpecific: boolean;
  };

The push apply loop now has a dedicated branch for it that calls
wmill.updateResource / updateVariable with just the ws_specific flag.
prettyChanges renders it on its own line. The dry-run JSON output picks
it up via the existing change.name / change.path passthrough.

The defensive wsSpecificMatches check inside push{Resource,Variable} is
no longer needed (sync push doesn't go through them for flag-only
diffs) and is reverted.

* drop folders

* feat(cli): warn on remote ws_specific items missing from local config

When 'wmill sync pull' fetches the server's ws_specific list, items the
server marks as ws_specific but that aren't matched by the local
wmill.yaml's specificItems patterns now produce a warning. The merge
already preserves correctness (those items are still treated as ws_specific
during this pull), but the user's config drifts from the remote — and a
later push from another machine without that config would push the item
as non-ws_specific. Surface the drift so the user can update wmill.yaml.

Also filter ws_specific_flag changes out before preCheckPermissionedAs
(it expects added/edited/deleted only and they have no content payload
so on_behalf_of resolution doesn't apply).

* fix(cli): scope ws_specific drift warning to items in this pull's changes

Previously the warning iterated every ws_specific item the server returned,
producing log spam for items unrelated to the current pull (items that
exist locally with no change, or items the user has nothing to do with
this round). Move the loop after compareDynFSElement and only warn for
items whose path appears in the changes list — i.e., items the user is
actually pulling right now.

* fix: clean up linked-side ws_specific rows on resource/variable delete

Three places left orphaned ws_specific rows behind:

1. delete_resource deleted the resource's own ws_specific row and the
   linked variables, but never the ws_specific 'variable' rows that
   mark_linked_variables_ws_specific had auto-inserted for those
   variable paths.
2. delete_variable deleted its own ws_specific row and the linked
   resource at the same path, but never a ws_specific 'resource' row at
   that path.
3. delete_resources_bulk didn't even cascade to linked variables, let
   alone clean up their ws_specific rows.

A new resource or variable later created at one of those paths would
silently inherit a stale ws_specific flag — list_ws_specific would
report it as workspace-specific, workspace diffs would treat it as
'no changes', and CLI sync would skip it.

Fix:

- delete_resource: DELETE FROM ws_specific WHERE item_kind = 'variable'
  AND path = ANY(linked_var_paths) before the linked-variable delete.
- delete_variable: DELETE FROM ws_specific WHERE item_kind = 'resource'
  AND path = path before the linked-resource delete.
- delete_resources_bulk: collect $var: refs from each bulk-deleted
  resource (mirror of single delete), then delete ws_specific 'variable'
  rows AND the variable rows themselves. Brings bulk delete in line with
  single delete semantics, including the orphan cleanup.

* fix: gate list_ws_specific by resource/variable RLS

The endpoint queried ws_specific directly under user_db, but ws_specific
itself has no per-item RLS — only a workspace-level column. Any workspace
member could enumerate every ws_specific path including those in folders
they lack read access to (e.g. f/finance/prod_db_creds), revealing path
existence that list_resources / list_variables would have hidden.

Add EXISTS clauses against resource and variable so the same path-based
RLS policies that govern those tables (see_own / see_member /
see_extra_perms_user / see_extra_perms_groups / see_folder_extra_perms_user)
also gate visibility here. The user transaction already establishes the
session context; the joins make the policies apply.

* only resources and variables

* fix(cli): make workspace-specific path mapping handle .json files

isSpecificItem() was extended to normalize .json -> .yaml so .json
files could be matched against patterns, but the surrounding helpers
remained yaml-only:

- toWorkspaceSpecificPath only mapped folder.meta.yaml / settings.yaml
  / .X.yaml — a foo.resource.json went through unchanged, so the
  workspace-specific filename was never produced.
- fromWorkspaceSpecificPath only matched .yaml extensions — pushing
  foo.dev.resource.json could not map back to foo.resource.json.
- isCurrentWorkspaceFile / isWorkspaceSpecificFile regexes ended in
  \.yaml$, missing every branch-specific .json file.

Replace the literal '.yaml' anchors with '(yaml|json)' alternations,
preserve the actual extension on round-trips, and rename the helper
buildYamlTypePattern -> buildItemTypePattern (it never had anything
extension-specific in it). getFileTypeSuffix now returns the matching
suffix for either extension. Changed:

- getFileTypeSuffix
- toWorkspaceSpecificPath / fromWorkspaceSpecificPath
- isCurrentWorkspaceFile / isWorkspaceSpecificFile
- isTriggerFile / isScheduleFile

isItemTypeConfigured / isSpecificItem don't need touching — their
checks run after normalizeJsonToYaml(), which already collapses both
extensions to .yaml at the entry.

* fix: create_resource?update_if_exists=true honors ws_specific=false

The upsert path matched on `unwrap_or(false)`, so an explicit
`ws_specific: false` and an absent flag were indistinguishable — both
fell through with no DELETE on the existing ws_specific row. Callers
trying to clear the flag via PUT-with-update_if_exists silently saw
their request ignored.

Mirror update_resource's three-way handling:

  Some(true)  -> INSERT (+ mark linked variables)
  Some(false) -> DELETE (only when update_if_exists, since a pure
                 create has no existing row anyway)
  None        -> leave the existing flag alone

create_variable doesn't have an upsert path (no ON CONFLICT), so the
same bug doesn't apply there.

* sqlx prepare

* test: cover ws_specific cleanup, RLS filtering, upsert clearing, and CLI .json paths

Backend (backend/tests/ws_specific.rs + fixture):

- test_linked_delete_cleanup: creates a ws_specific resource that
  references a variable via $var:, deletes the resource, asserts the
  cross-kind ws_specific row for the auto-marked variable is also
  removed. Then does the inverse for delete_variable, verifying the
  ws_specific 'resource' row at the same path is cleaned by variable
  delete.
- test_list_ws_specific_filters_by_rls: admin creates ws_specific items
  in u/test-user/ and u/test-user-2/; verifies admin sees both via
  list_ws_specific while a non-admin (test-user-2) only sees their own
  path — the RLS see_own policy on the joined resource/variable tables
  hides the other.
- test_create_resource_upsert_clears_ws_specific: covers the three-way
  Option<bool> handling on the upsert path: Some(true) inserts,
  Some(false) clears the existing row, None leaves it alone.

CLI:

- specific_items_unit.test.ts: add 14 tests covering toWorkspaceSpecificPath
  / fromWorkspaceSpecificPath / isWorkspaceSpecificFile /
  isCurrentWorkspaceFile / isSpecificItem / isItemTypeConfigured for
  .json files (variable, resource, trigger, schedule, folder.meta,
  settings).
- ws_specific_flag_only_unit.test.ts (new): covers
  computeWsSpecificFlagOnlyPushes — emits flag-only changes only for
  resource and variable kinds (the backend's list_ws_specific_versions
  scope), does not emit for schedules or triggers, returns empty when
  serverItems is null (older server), respects existing server entries,
  preserves .json extension on filePath.
- Export computeWsSpecificFlagOnlyPushes so it can be unit-tested.

* perf: index workspace_settings.deploy_to for the recursive CTE

list_ws_specific_versions's recursive CTE probes WHERE ws.deploy_to =
r.ws_id every iteration; without an index on workspace_settings.deploy_to
each iteration seq-scans the table — at 10M workspaces with the depth
cap of 32 that's up to 320M row reads per call.

deploy_to is sparse (most workspaces don't deploy anywhere), so a
partial index WHERE deploy_to IS NOT NULL stays small while still
covering every probe. Tucked into the existing migration since the
function and the index ship together.

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-05-07 13:35:54 +00:00

1101 lines
37 KiB
Rust

#[cfg(feature = "async")]
use futures::FutureExt;
use windmill_api::{
apis::{
configuration::Configuration,
job_api,
resource_api::{self, get_resource_value_interpolated},
variable_api,
},
models::{EditResource, GetCompletedJobResultMaybe200Response},
};
use serde::Deserialize;
use serde_json::{Value, json};
use std::{
env::{self, var},
fmt::Debug,
time::{Duration, Instant},
};
use crate::{maybe_future::maybe_future::MaybeFuture, ret};
#[derive(Debug, Clone)]
pub struct Windmill {
pub workspace: String,
pub token: String,
pub base_internal_url: String,
pub client_config: Configuration,
}
impl Windmill {
/// Creates a new `Windmill` instance with default configuration.
///
/// Reads configuration from environment variables:
/// - `WM_TOKEN` for authentication token
/// - `WM_WORKSPACE` for workspace name
/// - `BASE_INTERNAL_URL` for API base URL (appends `/api` automatically)
///
/// # Errors
/// Returns `SdkError` if:
/// - Required environment variables are missing
/// - Environment variables cannot be read
pub fn default() -> Result<Self, SdkError> {
Windmill::new(None, None, None)
}
/// Creates a new `Windmill` instance with optional overrides.
///
/// Falls back to environment variables for any `None` parameters:
/// - `WM_TOKEN` if `token` is `None`
/// - `WM_WORKSPACE` if `workspace` is `None`
/// - `BASE_INTERNAL_URL` + "/api" if `base_internal_url` is `None`
///
/// # Parameters
/// - `token`: Optional bearer token override
/// - `workspace`: Optional workspace name override
/// - `base_internal_url`: Optional base URL override (without `/api` suffix)
///
/// # Errors
/// Returns `SdkError` if:
/// - Required environment variables are missing when needed
/// - Environment variables cannot be read
pub fn new(
token: Option<String>,
workspace: Option<String>,
base_internal_url: Option<String>,
) -> Result<Self, SdkError> {
use env::var;
let (token, base_internal_url, workspace) = (
token.or(var("WM_TOKEN").ok()).ok_or(SdkError::BadValue(
"WM_TOKEN is not set nor was provided in constructor".to_owned(),
))?,
base_internal_url
.or(var("BASE_INTERNAL_URL").ok())
.ok_or(SdkError::BadValue(
"BASE_INTERNAL_URL is not set nor was provided in constructor".to_owned(),
))?
+ "/api",
workspace
.or(var("WM_WORKSPACE").ok())
.ok_or(SdkError::BadValue(
"WM_WORKSPACE is not set nor was provided in constructor".to_owned(),
))?,
);
Ok(Windmill {
client_config: Configuration {
// TODO: client: reqwest::blocking::Client::new(), // Use blocking client?
base_path: base_internal_url.clone(),
bearer_access_token: Some(token.clone()),
..Default::default()
},
workspace,
token,
base_internal_url,
})
}
/// Retrieves a variable from Windmill, automatically parsing it as JSON/YAML.
///
/// This is the convenience version that attempts to parse the variable value as:
/// 1. JSON (primary attempt)
/// 2. YAML (fallback if JSON parsing fails)
/// 3. Raw string (final fallback if both parsings fail)
///
/// For better performance when you know the format or don't need parsing,
/// use [`Self::get_variable_raw`] instead.
///
/// # Arguments
/// * `path` - Variable path (e.g., "u/user/variable_name")
///
/// # Example
/// ```no_run
///
/// use wmill::Windmill;
/// use serde_json::json;
///
/// let wm = Windmill::default()?;
///
/// // For a variable containing JSON: {"key": "value"}
/// let json_var = wm.get_variable("u/admin/config")?;
///
/// // For a variable containing plain text
/// let text_var = wm.get_variable("u/user/plaintext_note")?;
///
/// # Ok::<(), wmill::SdkError>(())
/// ```
///
/// See also: [`Self::get_variable_raw`] for the unparsed version
pub fn get_variable<'a>(&'a self, path: &'a str) -> MaybeFuture<'a, Result<Value, SdkError>> {
ret!(self.get_variable_inner(path));
}
async fn get_variable_inner<'a>(&'a self, path: &'a str) -> Result<Value, SdkError> {
let raw = self.get_variable_raw_inner(path).await?;
Ok(serde_json::from_str(&raw)
.or(serde_yaml::from_str(&raw))
.unwrap_or(json!(raw)))
}
/// This is the **faster version** when:
/// - You know the variable contains plain text
/// - You want to handle parsing yourself
/// - You need maximum performance
///
/// Performance benefit comes from avoiding:
/// 1. JSON parsing attempt
/// 2. YAML parsing fallback
///
/// # Arguments
/// * `path` - Variable path (e.g., "u/user/variable_name")
///
/// # Example
/// ```no_run
/// use wmill::Windmill;
/// use serde_json::{json, Value};
///
/// let wm = Windmill::default()?;
///
/// // When you need the raw content
/// let raw_content = wm.get_variable_raw("u/user/custom_format")?;
///
/// // When you know it's JSON and want to parse it differently
/// let json_value: Value = serde_json::from_str(
/// &wm.get_variable_raw("u/admin/config")?
/// )?;
///
/// # Ok::<(), wmill::SdkError>(())
/// ```
///
/// See also: [`Self::get_variable`] for the auto-parsed version
pub fn get_variable_raw<'a>(
&'a self,
path: &'a str,
) -> MaybeFuture<'a, Result<String, SdkError>> {
ret!(self.get_variable_raw_inner(path));
}
async fn get_variable_raw_inner<'a>(&'a self, path: &'a str) -> Result<String, SdkError> {
let v =
variable_api::get_variable_value(&self.client_config, &self.workspace, path, None).await?;
v.get(1..v.len() - 1)
.ok_or(SdkError::InternalError(
"returned value is incorrect".into(),
))
.map(|s| s.to_owned())
}
/// Creates or updates a variable in the workspace.
///
/// This function provides atomic variable management that:
/// - Creates a new variable if it doesn't exist
/// - Updates an existing variable if found
/// - Handles both regular and secret variables
///
/// # Parameters
/// - `value`: The variable value to set
/// - `path`: The variable path/identifier
/// - `is_secret`: Whether to store as a secret (encrypted) variable
///
/// # Errors
/// Returns `SdkError` if:
/// - Variable fetch fails for reasons other than "not found"
/// - Variable creation fails
/// - Variable update fails
/// - Underlying API calls fail
///
/// # Notes
/// - For new variables, defaults to empty description
/// - Updates only modify the value (preserving other metadata)
/// - Secret status can only be set during creation
pub fn set_variable<'a>(
&'a self,
value: String,
path: &'a str,
is_secret: bool,
) -> MaybeFuture<'a, Result<(), SdkError>> {
ret!(self.set_variable_inner(value, path, is_secret));
}
async fn set_variable_inner<'a>(
&'a self,
value: String,
path: &'a str,
is_secret: bool,
) -> Result<(), SdkError> {
let res =
variable_api::get_variable(&self.client_config, &self.workspace, path, None, None)
.await;
if res.is_err() {
variable_api::create_variable(
&self.client_config,
&self.workspace,
windmill_api::models::CreateVariable {
path: path.to_owned(),
value,
is_secret,
description: "".to_owned(),
account: None,
is_oauth: None,
expires_at: None,
labels: None,
ws_specific: None,
},
None,
)
.await?;
} else {
variable_api::update_variable(
&self.client_config,
&self.workspace,
path,
windmill_api::models::EditVariable {
path: None,
value: Some(value),
is_secret: None,
description: None,
labels: None,
ws_specific: None,
},
None,
)
.await?;
}
Ok(())
}
/// Fetches and deserializes a resource into a concrete type.
///
/// This is the recommended way to access resources when you know the expected type.
/// For raw JSON access or dynamic typing, use [`Self::get_resource_any`] instead.
///
/// # Type Parameters
/// * `T` - Any type implementing `DeserializeOwned` (most structs with `#[derive(Deserialize)]`)
///
/// # Arguments
/// * `path` - The resource path (e.g., "u/user/resource_name")
///
/// # Example
/// ```no_run
/// use wmill::Windmill;
/// use serde_json::{json, Value};
/// use serde::Deserialize;
///
/// #[derive(Deserialize)]
/// struct DbConfig {
/// url: String,
/// pool_size: Option<u32>,
/// }
///
/// let wm = Windmill::default()?;
///
/// // Directly deserialize to your type
/// let config: DbConfig = wm.get_resource("u/admin/db_connection")?;
/// # Ok::<(), wmill::SdkError>(())
/// ```
///
/// See also: [`Self::get_resource_any`] for untyped version
pub fn get_resource<'a, T: serde::de::DeserializeOwned>(
&'a self,
path: &'a str,
) -> MaybeFuture<'a, Result<T, SdkError>> {
ret!(self.get_resource_inner(path));
}
async fn get_resource_inner<'a, T: serde::de::DeserializeOwned>(
&'a self,
path: &'a str,
) -> Result<T, SdkError> {
Ok(serde_json::from_value(
self.get_resource_any_inner(path).await?,
)?)
}
/// Fetches a raw JSON [`Value`] from Windmill by path.
///
/// Use this when you need the raw JSON structure or don't have a concrete type to deserialize into.
/// For typed deserialization, prefer [`Self::get_resource`] instead.
///
/// # Arguments
/// * `path` - The resource path (e.g., "u/user/resource_name")
///
/// # Example
/// ```no_run
/// use wmill::Windmill;
///
/// let wm = Windmill::default()?;
///
/// // When you need to inspect the raw structure first
/// let json = wm.get_resource_any("u/admin/db_connection")?;
///
/// println!("Url is: {}", json["url"]);
///
/// # Ok::<(), wmill::SdkError>(())
/// ```
///
/// See also: [`Self::get_resource`] for typed version
pub fn get_resource_any<'a>(
&'a self,
path: &'a str,
) -> MaybeFuture<'a, Result<Value, SdkError>> {
ret!(self.get_resource_any_inner(path));
}
async fn get_resource_any_inner<'a>(&'a self, path: &'a str) -> Result<Value, SdkError> {
Ok(
get_resource_value_interpolated(&self.client_config, &self.workspace, path, None, None)
.await?,
)
}
/// Creates or updates a resource in Windmill.
///
/// This function sets a resource's value at the specified path, creating it if it doesn't exist
/// or updating it if it does. The resource will be of the specified type.
///
/// # Arguments
/// * `value` - The value to set for the resource. Use `None` to create an empty resource.
/// * `path` - The ownership path of the resource (e.g., "u/user/resource_name").
/// Defines permissions based on Windmill's path prefix system.
/// * `resource_type` - The type of resource to create (e.g., "postgresql", "smtp").
/// Must be a pre-existing resource type.
///
/// # Examples
/// ```no_run
/// use wmill::Windmill;
///
/// # fn main() -> anyhow::Result<()> {
/// let wm = Windmill::default()?;
/// wm.set_resource(
/// Some(serde_json::json!({"host": "localhost", "port": 5432})),
/// "u/admin/database",
/// "postgresql"
/// )?;
/// # Ok(())
/// # }
/// ```
pub fn set_resource<'a>(
&'a self,
value: Option<Value>,
path: &'a str,
resource_type: &'a str,
) -> MaybeFuture<'a, Result<(), SdkError>> {
ret!(self.set_resource_inner(value, path, resource_type));
}
async fn set_resource_inner<'a>(
&'a self,
value: Option<Value>,
path: &'a str,
resource_type: &'a str,
) -> Result<(), SdkError> {
// if resource_api::get_resource(&self.client_config, &self.workspace, path)
// .await
// .is_err()
if !resource_api::exists_resource(&self.client_config, &self.workspace, path).await? {
resource_api::create_resource(
&self.client_config,
&self.workspace,
windmill_api::models::CreateResource {
path: path.to_owned(),
value,
description: None,
resource_type: resource_type.to_owned(),
labels: None,
ws_specific: None,
},
// Some(true),
None,
)
.await?;
} else {
resource_api::update_resource(
&self.client_config,
&self.workspace,
path,
EditResource {
path: None,
description: None,
value: Some(value),
resource_type: Some(resource_type.to_owned()),
labels: None,
ws_specific: None,
},
)
.await?;
}
Ok(())
}
/// Retrieves and deserializes the current typed state value for a script's execution context.
///
/// This is the typed version of [`Self::get_state_any`], automatically deserializing the state
/// into the specified type `T` that implements [`serde::de::DeserializeOwned`].
///
/// # Type Parameter
/// * `T` - The type to deserialize into (must implement `serde::de::DeserializeOwned`)
///
/// # Examples
/// ```no_run
/// use wmill::Windmill;
/// use serde_json::{json, Value};
/// use serde::Deserialize;
///
/// #[derive(Deserialize)]
/// struct ScriptState {
/// counter: i32,
/// last_run: String,
/// }
///
/// let wm = Windmill::default()?;
///
/// // Get typed state
/// let state: ScriptState = wm.get_state()?;
///
/// println!("Counter: {}, Last run: {}", state.counter, state.last_run);
///
/// # Ok::<(), wmill::SdkError>(())
/// ```
///
/// # Behavior Details
/// - Uses same state path resolution as [`Self::get_state_any`] (`WM_STATE_PATH_NEW` → `WM_STATE_PATH` fallback)
/// - Performs runtime type checking during deserialization
///
/// # When to Use vs [`Self::get_state_any`]
/// | Use Case | `get_state<T>` | `get_state_any` |
/// |------------------------|-------------------------|------------------------|
/// | Known state structure | ✅ Preferred | ⚠️ Requires manual parsing |
/// | Dynamic state | ❌ Won't compile | ✅ Works |
/// | Type safety | ✅ Compile-time checks | ❌ Runtime checks only |
///
/// # Notes
/// - For complex types, derive `Deserialize` using Serde attributes
/// - Prefer this over [`Self::get_state_any`] when state schema is stable
/// - State modifications should use corresponding [`Self::set_state`] with matching type
///
/// # See Also
/// - [`Self::get_state_any`] - Untyped state access
/// - [`Self::set_state`] - For updating typed states
/// - [Windmill State Management](https://www.windmill.dev/docs/core_concepts/persistent_storage/within_windmill#states)
pub fn get_state<'a, T: serde::de::DeserializeOwned>(
&'a self,
) -> MaybeFuture<'a, Result<T, SdkError>> {
ret!(self.get_resource_inner(&get_state_path()?));
}
/// Retrieves and deserializes the typed state value from a custom resource path.
///
/// This is the same as [`Self::get_state`] but allows overriding the underlying state
/// resource path. Useful when you want to read/write a different state resource.
pub fn get_state_at<'a, T: serde::de::DeserializeOwned>(
&'a self,
path: &'a str,
) -> MaybeFuture<'a, Result<T, SdkError>> {
ret!(self.get_resource_inner(path));
}
/// Retrieves the current state value for a script's execution context.
///
/// States persist data between runs of the same script by the same trigger (schedule or user).
/// This is the untyped version that returns a generic [`Value`].
///
/// # Examples
/// ```no_run
/// use wmill::Windmill;
/// use serde_json::Value;
/// use serde::Deserialize;
///
/// let wm = Windmill::default()?;
///
/// // Get state (returns serde_json::Value)
/// let state: Value = wm.get_state_any()?;
///
/// // Use with default if empty
/// let counter = state.as_i64().unwrap_or(0);
/// # Ok::<(), wmill::SdkError>(())
/// ```
///
/// # Behavior Details
/// - Automatically uses the script's state path from `WM_STATE_PATH_NEW` (falls back to `WM_STATE_PATH`)
/// - Returns the full state object stored as a Windmill resource
/// - State resources are hidden from Workspace view but visible under Resources → States
///
/// # Typical Use Cases
/// 1. Maintaining counters between runs
/// 2. Storing last execution timestamps
/// 3. Keeping reference data (like previous API responses)
///
/// # Notes
/// - For typed state access, use `get_state<T>` instead
/// - States are isolated per script and trigger combination
/// - Maximum state size: 5MB (compressed)
///
/// # See Also
/// - [`Self::set_state`] - For updating the state
/// - [Windmill State Management](https://www.windmill.dev/docs/core_concepts/persistent_storage/within_windmill#states)
pub fn get_state_any<'a>(&'a self) -> MaybeFuture<'a, Result<Value, SdkError>> {
ret!(self.get_resource_any_inner(&get_state_path()?));
}
/// Retrieves the untyped state value from a custom resource path.
///
/// This is the same as [`Self::get_state_any`] but allows overriding the underlying state
/// resource path.
pub fn get_state_any_at<'a>(&'a self, path: &'a str) -> MaybeFuture<'a, Result<Value, SdkError>> {
ret!(self.get_resource_any_inner(path));
}
/// Updates or clears the script's persistent state.
///
/// # Arguments
/// * `value` - New state value (`Some(Value)`) or `None` to clear state
///
/// # Examples
/// ```no_run
/// use wmill::Windmill;
/// use serde_json::json;
///
/// # async fn example() -> anyhow::Result<()> {
/// let wm = Windmill::default()?;
///
/// // Set state
/// wm.set_state(Some(json!({"count": 42})))?;
///
/// // Clear state
/// wm.set_state(None)?;
/// # Ok(())
/// # }
/// ```
///
/// See also: [`Self::get_state`], [`Self::get_state_any`]
pub fn set_state<'a>(&'a self, value: Option<Value>) -> MaybeFuture<'a, Result<(), SdkError>> {
ret!(self.set_resource_inner(value, &get_state_path()?, "state"));
}
/// Updates or clears a state resource at a custom path.
///
/// This is the same as [`Self::set_state`] but allows overriding the target state resource
/// path.
pub fn set_state_at<'a>(
&'a self,
value: Option<Value>,
path: &'a str,
) -> MaybeFuture<'a, Result<(), SdkError>> {
ret!(self.set_resource_inner(value, path, "state"));
}
/// Executes a script synchronously and waits for its completion.
///
/// This is a blocking version of `run_script_async` that handles the entire script execution
/// lifecycle including job scheduling and result waiting.
///
/// # Parameters
/// - `ident`: Script identifier (either path or hash)
/// - `ident_is_hash`: Whether the identifier is a hash (true) or path (false)
/// - `args`: JSON arguments to pass to the script
/// - `scheduled_in_secs`: Optional delay before execution (in seconds)
/// - `timeout_secs`: Maximum time to wait for job completion (in seconds)
/// - `verbose`: Whether to print execution details
/// - `assert_result_is_not_none`: Whether to fail if the result is None
///
/// # Errors
/// Returns `SdkError` if:
/// - Script fails to start
/// - Job times out
/// - Result assertion fails
/// - Underlying API calls fail
pub fn run_script_sync<'a>(
&'a self,
ident: &'a str,
ident_is_hash: bool,
args: Value,
scheduled_in_secs: Option<u32>,
timeout_secs: Option<u64>,
verbose: bool,
assert_result_is_not_none: bool,
) -> MaybeFuture<'a, Result<Value, SdkError>> {
if verbose {
println!("running `{ident}` synchronously with {:?}", &args);
}
ret!(async move {
let job_id = self
.run_script_async_inner(ident, ident_is_hash, args, scheduled_in_secs)
.await?;
self.wait_job_inner(
&job_id.to_string(),
timeout_secs,
verbose,
assert_result_is_not_none,
)
.await
});
}
/// Asynchronously executes a script in Windmill and returns its job UUID.
///
/// This function runs a script either by path or by hash, with optional:
/// - Parent job inheritance (when run within a flow)
/// - Scheduled execution delay
/// - Argument passing
///
/// # Arguments
/// * `ident` - Script identifier (path or hash depending on `ident_is_hash`)
/// * `ident_is_hash` - If true, `ident` is treated as a script hash; if false, as a path
/// * `args` - JSON arguments to pass to the script (must be an object if using path)
/// * `scheduled_in_secs` - Optional delay (in seconds) before execution
///
/// # Examples
/// ```no_run
/// use wmill::Windmill;
/// use serde_json::json;
///
/// let wm = Windmill::default()?;
/// let job_id = wm.run_script_async(
/// "u/user/script_path",
/// false,
/// json!({"param1": "value1"}),
/// Some(10) // Run after 10 seconds
/// )?;
///
/// # Ok::<(), wmill::SdkError>(())
/// ```
///
/// # Behavior Details
/// - **Automatic Job Inheritance**:
/// - Detects `WM_JOB_ID` env var → sets as `parent_job`
/// - Detects `WM_ROOT_FLOW_JOB_ID` env var → sets as `root_job`
/// - **Scheduled Execution**:
/// - When `scheduled_in_secs` is provided, sets `scheduled_in_secs` in args
/// - **Argument Handling**:
/// - For path-based execution (`ident_is_hash=false`), args must be a JSON object
/// - For hash-based execution, args can be any valid JSON value
///
/// # Errors
/// - [`SdkError::BadValue`] if path-based execution receives non-object arguments
/// - API errors from Windmill's backend
pub fn run_script_async<'a>(
&'a self,
ident: &'a str,
ident_is_hash: bool,
args: Value,
scheduled_in_secs: Option<u32>,
) -> MaybeFuture<'a, Result<uuid::Uuid, SdkError>> {
ret!(self.run_script_async_inner(ident, ident_is_hash, args, scheduled_in_secs));
}
async fn run_script_async_inner<'a>(
&'a self,
ident: &'a str,
ident_is_hash: bool,
mut args: Value,
scheduled_in_secs: Option<u32>,
) -> Result<uuid::Uuid, SdkError> {
if let Ok(parent_job) = var("WM_JOB_ID") {
args["parent_job"] = json!(parent_job);
}
if let Ok(root_job) = var("WM_ROOT_FLOW_JOB_ID") {
args["root_job"] = json!(root_job);
}
if let Some(scheduled_in_secs) = scheduled_in_secs {
args["scheduled_in_secs"] = json!(scheduled_in_secs);
}
let uuid = if ident_is_hash {
job_api::run_script_by_hash(
&self.client_config,
&self.workspace,
ident,
args,
None,
None,
None,
None,
None,
None,
None,
None,
None,
)
.await?
} else {
job_api::run_script_by_path(
&self.client_config,
&self.workspace,
ident,
args.as_object()
.ok_or(SdkError::BadValue(format!(
"Args should be Object, but it is: {}",
args
)))?
.clone()
.into_iter()
.collect(),
None,
None,
None,
None,
None,
None,
None,
None,
)
.await?
};
Ok(uuid)
}
/// Waits for a job to complete and returns its result.
///
/// This function provides both synchronous and asynchronous interfaces for waiting
/// on job completion, with timeout handling and result validation.
///
/// # Parameters
/// - `job_id`: The ID of the job to wait for
/// - `timeout_secs`: Maximum time to wait (in seconds) before cancelling the job
/// - `verbose`: Whether to print progress information
/// - `assert_result_is_not_none`: Whether to fail if the job returns no result
///
/// # Errors
/// Returns `SdkError` if:
/// - Job fails or times out
/// - Result assertion fails when `assert_result_is_not_none` is true
/// - Underlying API calls fail
///
/// # Behavior
/// 1. Polls job status at 500ms intervals
/// 2. Cancels job if timeout is reached
/// 3. Validates success status and optional result presence
/// 4. Returns either the result or appropriate error
pub fn wait_job<'a>(
&'a self,
job_id: &'a str,
timeout_secs: Option<u64>,
verbose: bool,
assert_result_is_not_none: bool,
) -> MaybeFuture<'a, Result<Value, SdkError>> {
ret!(self.wait_job_inner(job_id, timeout_secs, verbose, assert_result_is_not_none));
}
async fn wait_job_inner<'a>(
&'a self,
job_id: &'a str,
timeout_secs: Option<u64>,
verbose: bool,
assert_result_is_not_none: bool,
) -> Result<Value, SdkError> {
let start = Instant::now();
loop {
if let Some(timeout) = timeout_secs {
if start.elapsed().as_secs() > timeout {
println!("WARN: reached timeout for {job_id}. Cancelling the job.");
job_api::cancel_queued_job(
&self.client_config,
&self.workspace,
job_id,
windmill_api::models::CancelQueuedJobRequest {
reason: Some("reached timeout".into()),
},
)
.await?;
}
}
let GetCompletedJobResultMaybe200Response {
completed,
result,
success,
started,
} = job_api::get_completed_job_result_maybe(
&self.client_config,
&self.workspace,
job_id,
Some(true),
)
.await?;
if matches!(started, Some(false)) && verbose {
println!("INFO: job {job_id} has not started yet");
}
if completed {
if matches!(success, Some(true)) {
if result.is_none() && assert_result_is_not_none {
return Err(SdkError::ExecutionError("Result was None".into()));
}
return Ok(result.unwrap_or_default());
} else {
return Err(SdkError::ExecutionError(format!(
"Job {job_id} was not successful: {:?}",
result
)));
}
}
if verbose {
println!("INFO: sleeping 0.5 seconds for {job_id}");
}
tokio::time::sleep(Duration::from_millis(500)).await;
}
}
/// Retrieves the current status of a Windmill job by its UUID.
///
/// This function queries the Windmill backend to determine whether a job is:
/// - Waiting to be executed
/// - Currently running
/// - Already completed
///
/// # Arguments
/// * `job_id` - The UUID of the job to check (format: "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx")
///
/// # Job Lifecycle States
/// - **Waiting**: Job is queued but hasn't started execution
/// - **Running**: Job is currently being executed
/// - **Completed**: Job has finished (successfully or with errors)
pub fn get_job_status<'a>(
&'a self,
job_id: &'a str,
) -> MaybeFuture<'a, Result<JobStatus, SdkError>> {
ret!(self.get_job_status_inner(job_id));
}
async fn get_job_status_inner<'a>(&'a self, job_id: &'a str) -> Result<JobStatus, SdkError> {
let job = job_api::get_job(
&self.client_config,
&self.workspace,
job_id,
Some(true),
Some(true),
None,
)
.await?;
Ok(match job {
windmill_api::models::Job::JobOneOf(..) => JobStatus::Completed,
windmill_api::models::Job::JobOneOf1(job_one_of1) => {
if job_one_of1.running {
JobStatus::Running
} else {
JobStatus::Waiting
}
}
})
}
/// Retrieves the result of a completed job.
///
/// This provides both synchronous and asynchronous interfaces for fetching
/// the final result of a successfully completed job.
///
/// # Parameters
/// - `job_id`: The ID of the completed job to fetch results for
///
/// # Errors
/// Returns `SdkError` if:
/// - Job is not found
/// - Job failed to complete successfully
/// - Underlying API calls fail
/// - Result cannot be parsed
///
/// # Notes
/// - Only works for jobs that have already completed
/// - For pending/running jobs, use `wait_job` instead
/// - Does not handle timeouts or polling - assumes job is already complete
pub fn get_result<'a>(&'a self, job_id: &'a str) -> MaybeFuture<'a, Result<Value, SdkError>> {
ret!(self.get_result_inner(job_id));
}
async fn get_result_inner<'a>(&'a self, job_id: &'a str) -> Result<Value, SdkError> {
Ok(job_api::get_completed_job_result(
&self.client_config,
&self.workspace,
job_id,
None,
None,
None,
None,
)
.await?)
}
/// Updates the progress percentage of a running Windmill job.
///
/// This function allows scripts to report their execution progress (0-100%) back to the Windmill UI.
/// Progress updates are visible in both the jobs dashboard and flow visualizations.
///
/// # Arguments
/// * `value` - Progress percentage (0-100)
/// * `job_id` - Optional job UUID. If None, uses current job's ID from `WM_JOB_ID` environment variable
///
/// # Examples
/// ```no_run
/// use wmill::Windmill;
/// # fn main () -> anyhow::Result<()>{
/// let wm = Windmill::default()?;
///
/// // Report progress for current job
/// wm.set_progress(25, None)?;
/// # Ok(())
/// # }
///
/// ```
///
/// # Behavior Details
/// - Automatically handles flow context by detecting parent job ID
/// - Progress updates are reflected in real-time in the Windmill UI
/// - Typical usage pattern:
/// ```ignore
/// for (i, item) in items.iter().enumerate() {
/// process(item);
/// let progress = ((i + 1) * 100 / items.len()) as i32;
/// wmill.set_progress(progress, None).await?;
/// }
/// ```
///
/// # Notes
/// - Only affects jobs that are currently running
/// - Progress values outside 0-99 range are clamped by the server
/// - Progress cannot decrease
/// - For flows, updates the progress of both the step and parent flow
///
/// # See Also
/// - [Flow Progress Tracking](https://www.windmill.dev/docs/advanced/explicit_progress)
/// - [`Self::get_progress`] - For reading job progress
pub fn set_progress<'a>(
&'a self,
value: u32,
job_id: Option<String>,
) -> MaybeFuture<'a, Result<(), SdkError>> {
let f = async move {
let job_id = job_id.unwrap_or(var("WM_JOB_ID")?);
let job = job_api::get_job(
&self.client_config,
&self.workspace,
&job_id,
Some(true),
Some(true),
None,
)
.await?;
let flow_id = match job {
windmill_api::models::Job::JobOneOf(job) => job.parent_job,
windmill_api::models::Job::JobOneOf1(job) => job.parent_job,
};
windmill_api::apis::metrics_api::set_job_progress(
&self.client_config,
&self.workspace,
&job_id,
windmill_api::models::SetJobProgressRequest {
percent: Some(value as i32),
flow_job_id: flow_id,
},
)
.await?;
Ok(())
};
ret!(f);
}
/// Retrieves the current progress percentage of a Windmill job.
///
/// This function queries the Windmill backend to get the execution progress (0-100%)
/// of either a specific job or the current job context.
///
/// # Arguments
/// * `job_id` - Optional job UUID. If `None`, uses current job's ID from `WM_JOB_ID` env var
///
/// # See Also
/// - [Flow Progress Tracking](https://www.windmill.dev/docs/advanced/explicit_progress)
/// - [`Self::set_progress`] - For updating job progress
pub fn get_progress<'a>(
&'a self,
job_id: Option<String>,
) -> MaybeFuture<'a, Result<u32, SdkError>> {
let f = async move {
let job_id = job_id.unwrap_or(var("WM_JOB_ID")?);
Ok(windmill_api::apis::metrics_api::get_job_progress(
&self.client_config,
&self.workspace,
&job_id,
)
.await
.map(|v| v as u32)?)
};
ret!(f);
}
/// Executes an API call in either asynchronous or synchronous mode based on compilation context.
///
/// This function serves as a bridge between async and sync code, automatically adapting its behavior:
/// - With `async` feature: Returns a boxed future for later await
/// - Without `async` feature: Blocks immediately using the global runtime
///
/// # Examples
///
/// ## Async usage (with `async` feature)
/// ```ignore
/// use wmill::Windmill;
/// # #[tokio::main]
/// # async fn main() -> anyhow::Result<()>{
/// let wm = Windmill::default()?;
/// let user = wm.call_api(wmill::apis::admin_api::get_user(&wm.client_config, &wm.workspace, "Bob"))?;
/// println!("User details: {:?}", user);
/// # Ok(())
/// # }
/// ```
///
/// ## Sync usage (without `async` feature)
/// ```ignore
/// use wmill::Windmill;
/// let wm = Windmill::new(Some("<TOKEN>".into()), Some("admins".into()), Some("http://localhost:5000/api".into()))?;
/// let user = wm.call_api(wmill::apis::admin_api::get_user(&wm.client_config, &wm.workspace, "Bob"));
/// println!("User details: {:?}", user);
/// # Ok::<(), wmill::SdkError>(())
/// ```
pub fn call_api<'a, T>(
&'a self,
callback: impl Future<Output = T> + std::marker::Send + 'a,
) -> MaybeFuture<'a, T> {
ret!(callback);
}
// pub fn get_version() {}
}
pub enum JobStatus {
Running,
Waiting,
Completed,
}
fn get_state_path() -> Result<String, SdkError> {
Ok(var("WM_STATE_PATH_NEW").unwrap_or(var("WM_STATE_PATH")?))
}
#[derive(thiserror::Error, Debug)]
pub enum SdkError {
#[error("Error: {0}")]
Serde(#[from] serde_json::Error),
#[error("Having troubles reading Env Variable: {0}")]
VarError(#[from] std::env::VarError),
#[error("Api error: {0}")]
ApiError(String),
#[error("Internal Error: {0}")]
InternalError(String),
#[error("Specified value is incorrect: {0}")]
BadValue(String),
#[error("{0}")]
ExecutionError(String),
}
impl<T: for<'a> Deserialize<'a>> From<windmill_api::apis::Error<T>> for SdkError {
fn from(value: windmill_api::apis::Error<T>) -> Self {
Self::ApiError(value.to_string())
}
}