stormchaser-api 1.3.0

A robust, distributed workflow engine for event-driven and human-triggered workflows.
Documentation 
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

use serde_json::Value;
use sqlx::PgPool;
use stormchaser_model::CronWorkflowId;

use stormchaser_model::cron;

/// Creates a new cron workflow configuration.
#[allow(clippy::too_many_arguments)]
pub async fn create_cron_workflow(
    pool: &PgPool,
    id: CronWorkflowId,
    name: &str,
    description: &Option<String>,
    cronspec: &str,
    workflow_name: &str,
    repo_url: &str,
    workflow_path: &str,
    git_ref: &str,
    inputs: &Value,
    secret_token: &str,
    external_job_id: &Option<String>,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        r#"
        INSERT INTO cron_workflows (id, name, description, cronspec, workflow_name, repo_url, workflow_path, git_ref, inputs, secret_token, external_job_id)
        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
        "#,
    )
    .bind(id)
    .bind(name)
    .bind(description)
    .bind(cronspec)
    .bind(workflow_name)
    .bind(repo_url)
    .bind(workflow_path)
    .bind(git_ref)
    .bind(inputs)
    .bind(secret_token)
    .bind(external_job_id)
    .execute(pool)
    .await?;
    Ok(())
}

/// Retrieves all cron workflows.
/// List cron workflows.
pub async fn list_cron_workflows(pool: &PgPool) -> Result<Vec<cron::CronWorkflow>, sqlx::Error> {
    sqlx::query_as("SELECT * FROM cron_workflows ORDER BY created_at DESC")
        .fetch_all(pool)
        .await
}

/// Retrieves a cron workflow by ID.
/// Get cron workflow.
pub async fn get_cron_workflow(
    pool: &PgPool,
    id: CronWorkflowId,
) -> Result<Option<cron::CronWorkflow>, sqlx::Error> {
    sqlx::query_as("SELECT * FROM cron_workflows WHERE id = $1")
        .bind(id)
        .fetch_optional(pool)
        .await
}

/// Retrieves an active cron workflow by ID.
/// Get active cron workflow.
pub async fn get_active_cron_workflow(
    pool: &PgPool,
    id: CronWorkflowId,
) -> Result<Option<cron::CronWorkflow>, sqlx::Error> {
    sqlx::query_as("SELECT * FROM cron_workflows WHERE id = $1 AND is_active = TRUE")
        .bind(id)
        .fetch_optional(pool)
        .await
}

/// Deletes a scheduled workflow configuration
pub async fn delete_cron_workflow(pool: &PgPool, id: CronWorkflowId) -> Result<(), sqlx::Error> {
    sqlx::query("DELETE FROM cron_workflows WHERE id = $1")
        .bind(id)
        .execute(pool)
        .await?;
    Ok(())
}

/// Inserts a new cron workflow.
pub async fn insert_cron_workflow(
    pool: &PgPool,
    id: CronWorkflowId,
    payload: &crate::routes::CreateCronWorkflowRequest,
    secret_token: &str,
    external_job_id: Option<String>,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        r#"
        INSERT INTO cron_workflows (id, name, description, cronspec, workflow_name, repo_url, workflow_path, git_ref, inputs, secret_token, external_job_id)
        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
        "#,
    )
    .bind(id)
    .bind(&payload.name)
    .bind(&payload.description)
    .bind(&payload.cronspec)
    .bind(&payload.workflow_name)
    .bind(&payload.repo_url)
    .bind(&payload.workflow_path)
    .bind(&payload.git_ref)
    .bind(&payload.inputs)
    .bind(secret_token)
    .bind(external_job_id)
    .execute(pool)
    .await?;
    Ok(())
}