pipedash-web 0.3.0

REST API server for Pipedash
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

use axum::{
    extract::{
        Path,
        State,
    },
    routing::{
        delete,
        get,
    },
    Json,
    Router,
};
use serde::Serialize;

use crate::error::{
    ApiResult,
    AppError,
};
use crate::state::AppState;

#[derive(Debug, Serialize)]
pub struct CacheStats {
    pub pipelines_count: i64,
    pub run_history_count: i64,
    pub workflow_params_count: i64,
    pub metrics_count: i64,
}

#[derive(Debug, Serialize)]
pub struct CacheClearResponse {
    pub cleared: usize,
}

pub fn router() -> Router<AppState> {
    Router::new()
        .route("/stats", get(get_cache_stats))
        .route(
            "/run-history/{pipeline_id}",
            delete(clear_run_history_cache),
        )
        .route("/run-history", delete(clear_all_run_history_caches))
        .route("/pipelines", delete(clear_pipelines_cache))
        .route("/workflow-params", delete(clear_workflow_params_cache))
        .route("/", delete(clear_all_caches))
}

async fn get_cache_stats(State(state): State<AppState>) -> ApiResult<Json<CacheStats>> {
    let inner = state.inner.read().await;
    let core = inner.core.as_ref().ok_or_else(AppError::not_initialized)?;
    let pipelines_count = core
        .pipeline_service
        .get_pipelines_cache_count()
        .await
        .unwrap_or(0);

    let run_history_count = core
        .pipeline_service
        .get_run_history_cache_count()
        .await
        .unwrap_or(0);

    let workflow_params_count = core
        .pipeline_service
        .get_workflow_params_cache_count()
        .await
        .unwrap_or(0);

    let metrics_count = if let Some(metrics_service) = &core.metrics_service {
        metrics_service
            .get_storage_stats()
            .await
            .map(|stats| stats.total_metrics_count)
            .unwrap_or(0)
    } else {
        0
    };

    Ok(Json(CacheStats {
        pipelines_count,
        run_history_count,
        workflow_params_count,
        metrics_count,
    }))
}

async fn clear_run_history_cache(
    State(state): State<AppState>, Path(pipeline_id): Path<String>,
) -> ApiResult<()> {
    let inner = state.inner.read().await;
    let core = inner.core.as_ref().ok_or_else(AppError::not_initialized)?;
    core.pipeline_service
        .clear_run_history_cache(&pipeline_id)
        .await;
    Ok(())
}

async fn clear_all_run_history_caches(State(state): State<AppState>) -> ApiResult<()> {
    let inner = state.inner.read().await;
    let core = inner.core.as_ref().ok_or_else(AppError::not_initialized)?;
    core.pipeline_service.clear_all_run_history_caches().await;
    Ok(())
}

async fn clear_pipelines_cache(
    State(state): State<AppState>,
) -> ApiResult<Json<CacheClearResponse>> {
    let inner = state.inner.read().await;
    let core = inner.core.as_ref().ok_or_else(AppError::not_initialized)?;
    let cleared = core.pipeline_service.clear_pipelines_cache().await?;
    Ok(Json(CacheClearResponse { cleared }))
}

async fn clear_workflow_params_cache(State(state): State<AppState>) -> ApiResult<()> {
    let inner = state.inner.read().await;
    let core = inner.core.as_ref().ok_or_else(AppError::not_initialized)?;
    core.pipeline_service.clear_workflow_params_cache().await?;
    Ok(())
}

async fn clear_all_caches(State(state): State<AppState>) -> ApiResult<()> {
    let inner = state.inner.read().await;
    let core = inner.core.as_ref().ok_or_else(AppError::not_initialized)?;
    core.pipeline_service.clear_all_run_history_caches().await;
    core.pipeline_service.clear_pipelines_cache().await?;
    core.pipeline_service.clear_workflow_params_cache().await?;
    Ok(())
}