starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-08 08:04:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: payload capture of json to initialize flow input (#655)

Browse Source 
* capture table and some endpoints for json pastebin

* only retain user's most recent captures

* merge conflict

* sqlx data

* fix migration

Co-authored-by: sqwishy <somebody@froghat.ca>
This commit is contained in:
Ruben Fiszel
2022-10-25 00:54:53 +02:00
committed by GitHub
co-authored by sqwishy
parent 7addf0c6b1
commit 9a67607b20
6 changed files with 255 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/migrations/20220802211304_capture.down.sql
+1
View File
@@ -0,0 +1 @@
DROP TABLE capture;
backend/migrations/20220802211304_capture.up.sql
+21
View File
@@ -0,0 +1,21 @@
CREATE TABLE capture (
workspace_id VARCHAR(50) NOT NULL,
path VARCHAR(255) NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
created_by VARCHAR(50) NOT NULL,
payload JSONB NOT NULL DEFAULT 'null'::jsonb
CHECK (length(payload::text) < 10 * 1024),
PRIMARY KEY (workspace_id, path),
FOREIGN KEY (workspace_id) REFERENCES workspace(id)
);
ALTER TABLE capture ENABLE ROW LEVEL SECURITY;
CREATE POLICY see_own ON capture FOR ALL
USING ( SPLIT_PART(capture.path, '/', 1) = 'u'
AND SPLIT_PART(capture.path, '/', 2) = current_setting('session.user'));
CREATE POLICY see_member ON capture FOR ALL
USING ( SPLIT_PART(capture.path, '/', 1) = 'g'
AND SPLIT_PART(capture.path, '/', 2) = any(regexp_split_to_array(current_setting('session.groups'), ',')::text[]));
backend/openapi.yaml
+41
View File
@@ -3268,6 +3268,47 @@ paths:
text/plain:
schema:
type: string
/w/{workspace}/capture/{path}:
put:
summary: create flow preview capture
operationId: createCapture
tags:
- capture
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/Path"
responses:
"201":
description: flow preview capture created
post:
summary: update flow preview capture
operationId: updateCapture
tags:
- capture
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/Path"
responses:
"204":
description: flow preview captured
get:
summary: get flow preview capture
operationId: getCapture
tags:
- capture
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/Path"
responses:
"200":
description: captured flow preview
content:
application/json:
schema: {}
"404":
description: capture does not exist for this flow
components:
securitySchemes:
bearerAuth:
backend/sqlx-data.json
+63
View File
@@ -181,6 +181,20 @@
},
"query": "INSERT INTO workspace_invite\n (workspace_id, email, is_admin)\n VALUES ($1, $2, $3)"
},
"0a9a191273c735c41d56ea46a39ffca075a0550eada87df7162c5037164ad6bf": {
"describe": {
"columns": [],
"nullable": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Varchar"
]
}
},
"query": "\n INSERT INTO capture\n (workspace_id, path, created_by)\n VALUES ($1, $2, $3)\n ON CONFLICT (workspace_id, path)\n DO UPDATE SET created_at = now()\n "
},
"0ae9160591ae00117d20a616cfe07e38f0c32953c7e881e916c389255190b72d": {
"describe": {
"columns": [],
@@ -2714,6 +2728,20 @@
},
"query": "SELECT (flow_status->'step')::integer FROM queue WHERE id = $1"
},
"b9468b9e16f55db11b33d8e9793e6e3ae6c5add6ca02414140adb724120a6800": {
"describe": {
"columns": [],
"nullable": [],
"parameters": {
"Left": [
"Text",
"Text",
"Jsonb"
]
}
},
"query": "\n UPDATE capture\n SET payload = $3\n WHERE workspace_id = $1\n AND path = $2\n "
},
"bb56e61c7cfb09c0a28fb3226dfe91704c70d9fe15eda18e6889adfd7496f80b": {
"describe": {
"columns": [
@@ -3211,6 +3239,20 @@
},
"query": "SELECT EXISTS(SELECT 1 FROM workspace WHERE workspace.id = $1)"
},
"e02b99525cb1f8737acfec86809f59c8cff67bb1ec3926680fb691cc3573738a": {
"describe": {
"columns": [],
"nullable": [],
"parameters": {
"Left": [
"Text",
"Text",
"Int8"
]
}
},
"query": "\n DELETE FROM capture\n WHERE workspace_id = $1\n AND created_by = $2\n AND created_at <=\n ( SELECT created_at\n FROM capture\n WHERE workspace_id = $1\n AND created_by = $2\n ORDER BY created_at DESC\n OFFSET $3\n LIMIT 1 )\n "
},
"e262f83b672a558092dc959b28a919f18e44b2a0d03b27b0ddba99896940c6d3": {
"describe": {
"columns": [
@@ -3387,6 +3429,27 @@
},
"query": "DELETE FROM usr_to_group WHERE usr = $1 AND group_ = $2 AND workspace_id = $3"
},
"f4960efbb4189b595541608c9ab5cbbd56eb42059e961b6338b685105626a5c7": {
"describe": {
"columns": [
{
"name": "payload",
"ordinal": 0,
"type_info": "Jsonb"
}
],
"nullable": [
false
],
"parameters": {
"Left": [
"Text",
"Text"
]
}
},
"query": "\n SELECT payload\n FROM capture\n WHERE workspace_id = $1\n AND path = $2\n "
},
"f4a1bb4dfeafa3f28385c19d10730f2f49b4ebe5bba631cee46c760cee2e6949": {
"describe": {
"columns": [
backend/src/capture.rs
+125
View File
@@ -0,0 +1,125 @@
use axum::{
extract::{Extension, Path},
routing::{get, post, put},
Json, Router,
};
use hyper::StatusCode;
use crate::{
db::{UserDB, DB},
error::{JsonResult, Result},
users::Authed,
utils::{not_found_if_none, StripPath},
};
const KEEP_LAST: i64 = 8;
pub fn workspaced_service() -> Router {
Router::new()
.route("/*path", put(new_payload))
.route("/*path", get(get_payload))
}
pub fn global_service() -> Router {
Router::new().route("/*path", post(update_payload))
}
pub async fn new_payload(
authed: Authed,
Extension(user_db): Extension<UserDB>,
Path((w_id, path)): Path<(String, StripPath)>,
) -> Result<StatusCode> {
let mut tx = user_db.begin(&authed).await?;
sqlx::query!(
"
INSERT INTO capture
(workspace_id, path, created_by)
VALUES ($1, $2, $3)
ON CONFLICT (workspace_id, path)
DO UPDATE SET created_at = now()
",
&w_id,
&path.to_path(),
&authed.username,
)
.execute(&mut tx)
.await?;
/* Retain only KEEP_LAST most recent captures by this user in this workspace. */
sqlx::query!(
"
DELETE FROM capture
WHERE workspace_id = $1
AND created_by = $2
AND created_at <=
( SELECT created_at
FROM capture
WHERE workspace_id = $1
AND created_by = $2
ORDER BY created_at DESC
OFFSET $3
LIMIT 1 )
",
&w_id,
&authed.username,
KEEP_LAST,
)
.execute(&mut tx)
.await?;
tx.commit().await?;
Ok(StatusCode::CREATED)
}
pub async fn update_payload(
Extension(db): Extension<DB>,
Path((w_id, path)): Path<(String, StripPath)>,
Json(payload): Json<serde_json::Value>,
) -> Result<StatusCode> {
let mut tx = db.begin().await?;
sqlx::query!(
"
UPDATE capture
SET payload = $3
WHERE workspace_id = $1
AND path = $2
",
&w_id,
&path.to_path(),
&payload,
)
.execute(&mut tx)
.await?;
tx.commit().await?;
Ok(StatusCode::NO_CONTENT)
}
pub async fn get_payload(
authed: Authed,
Extension(user_db): Extension<UserDB>,
Path((w_id, path)): Path<(String, StripPath)>,
) -> JsonResult<serde_json::Value> {
let mut tx = user_db.begin(&authed).await?;
let payload = sqlx::query_scalar!(
"
SELECT payload
FROM capture
WHERE workspace_id = $1
AND path = $2
",
&w_id,
&path.to_path(),
)
.fetch_optional(&mut tx)
.await?;
tx.commit().await?;
not_found_if_none(payload, "capture", path.to_path()).map(axum::Json)
}
backend/src/lib.rs
+4 -1
View File
@@ -22,6 +22,7 @@ extern crate magic_crypt;
extern crate dotenv;
mod audit;
mod capture;
mod client;
mod db;
mod error;
@@ -155,7 +156,8 @@ pub async fn run_server(
.nest("/audit", audit::workspaced_service())
.nest("/acls", granular_acls::workspaced_service())
.nest("/workspaces", workspaces::workspaced_service())
.nest("/flows", flows::workspaced_service()),
.nest("/flows", flows::workspaced_service())
.nest("/capture", capture::workspaced_service()),
)
.nest("/workspaces", workspaces::global_service())
.nest(
@@ -169,6 +171,7 @@ pub async fn run_server(
.route_layer(from_extractor::<users::Authed>())
.route_layer(from_extractor::<users::Tokened>())
.nest("/w/:workspace_id/jobs", jobs::global_service())
.nest("/w/:workspace_id/capture", capture::global_service())
.nest(
"/auth",
users::make_unauthed_service().layer(Extension(argon2)),