11910714914

Committed 19 Nov 2024 10:06AM UTC coverage: 90.445% (-0.2%) from 90.687%

Build # 11910714914

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #994 from geo-engine/workspace-dependencies

use workspace dependencies, update toolchain, use global lock in expression

Run Details

9 of 11 new or added lines in 6 files covered. (81.82%)

375 existing lines in 75 files now uncovered.

132867 of 146904 relevant lines covered (90.44%)

54798.11 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.67

/services/src/pro/workflows/postgres_workflow_registry.rs

use crate::error::Result;
use crate::pro::contexts::ProPostgresDb;
use crate::workflows::workflow::{Workflow, WorkflowId};
use crate::{error, workflows::registry::WorkflowRegistry};
use async_trait::async_trait;
use bb8_postgres::{
    tokio_postgres::tls::MakeTlsConnect, tokio_postgres::tls::TlsConnect, tokio_postgres::Socket,
};
use snafu::ResultExt;

#[async_trait]
impl<Tls> WorkflowRegistry for ProPostgresDb<Tls>
where
    Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static + std::fmt::Debug,
    <Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
    <Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
    <<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
{
    async fn register_workflow(&self, workflow: Workflow) -> Result<WorkflowId> {
        let conn = self.conn_pool.get().await?;
        let stmt = conn
            .prepare(
                "INSERT INTO workflows (id, workflow) VALUES ($1, $2) 
            ON CONFLICT DO NOTHING;",
            )
            .await?;

        let workflow_id = WorkflowId::from_hash(&workflow);

        conn.execute(
            &stmt,
            &[
                &workflow_id,
                &serde_json::to_value(&workflow).context(error::SerdeJson)?,
            ],
        )
        .await?;

        Ok(workflow_id)
    }

    async fn load_workflow(&self, id: &WorkflowId) -> Result<Workflow> {
        // TODO: add authorization beyond the fact that you can only access workflows if you happen to know the id.
        let conn = self.conn_pool.get().await?;
        let stmt = conn
            .prepare("SELECT workflow FROM workflows WHERE id = $1")
            .await?;

        let row = conn.query(&stmt, &[&id]).await?;

        if row.is_empty() {
            return Err(error::Error::NoWorkflowForGivenId);
        }

        Ok(serde_json::from_value(row[0].get(0)).context(error::SerdeJson)?)
    }
}

1	use crate::error::Result;
2	use crate::pro::contexts::ProPostgresDb;
3	use crate::workflows::workflow::{Workflow, WorkflowId};
4	use crate::{error, workflows::registry::WorkflowRegistry};
5	use async_trait::async_trait;
6	use bb8_postgres::{
7	tokio_postgres::tls::MakeTlsConnect, tokio_postgres::tls::TlsConnect, tokio_postgres::Socket,
8	};
9	use snafu::ResultExt;
10
11	#[async_trait]
12	impl<Tls> WorkflowRegistry for ProPostgresDb<Tls>
13	where
14	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static + std::fmt::Debug,
15	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
16	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
17	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
18	{
19	async fn register_workflow(&self, workflow: Workflow) -> Result<WorkflowId> {	8✔
20	let conn = self.conn_pool.get().await?;	8✔
21	let stmt = conn	8✔
22	.prepare(	8✔
23	"INSERT INTO workflows (id, workflow) VALUES ($1, $2)	8✔
24	ON CONFLICT DO NOTHING;",	8✔
25	)	8✔
26	.await?;	8✔
27
28	let workflow_id = WorkflowId::from_hash(&workflow);	8✔
29		8✔
30	conn.execute(	8✔
31	&stmt,	8✔
32	&[	8✔
33	&workflow_id,	8✔
34	&serde_json::to_value(&workflow).context(error::SerdeJson)?,	8✔
35	],
36	)
37	.await?;	8✔
38
39	Ok(workflow_id)	8✔
40	}	16✔
41
42	async fn load_workflow(&self, id: &WorkflowId) -> Result<Workflow> {	8✔
43	// TODO: add authorization beyond the fact that you can only access workflows if you happen to know the id.
44	let conn = self.conn_pool.get().await?;	8✔
45	let stmt = conn	8✔
46	.prepare("SELECT workflow FROM workflows WHERE id = $1")	8✔
47	.await?;	8✔
48
49	let row = conn.query(&stmt, &[&id]).await?;	8✔
50
51	if row.is_empty() {	8✔
UNCOV 52	return Err(error::Error::NoWorkflowForGivenId);	×
53	}	8✔
54		8✔
55	Ok(serde_json::from_value(row[0].get(0)).context(error::SerdeJson)?)	8✔
56	}	16✔
57	}

geo-engine / geoengine / 11910714914

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous