5624216179

Committed 21 Jul 2023 03:05PM UTC coverage: 89.186% (+0.03%) from 89.156%

Build # 5624216179

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #831 from geo-engine/free-postgres

Free-postgres

Run Details

4197 of 4197 new or added lines in 44 files covered. (100.0%)

105664 of 118476 relevant lines covered (89.19%)

61107.62 hits per line

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

96.67

/services/src/workflows/postgres_workflow_registry.rs

use crate::contexts::PostgresDb;
use crate::error::Result;
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 PostgresDb<Tls>
where
    Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
    <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::contexts::PostgresDb;
2	use crate::error::Result;
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 PostgresDb<Tls>
13	where
14	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
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> {	4✔
20	let conn = self.conn_pool.get().await?;	4✔
21	let stmt = conn	4✔
22	.prepare(	4✔
23	"INSERT INTO workflows (id, workflow) VALUES ($1, $2)	4✔
24	ON CONFLICT DO NOTHING;",	4✔
25	)	4✔
26	.await?;	4✔
27
28	let workflow_id = WorkflowId::from_hash(&workflow);	4✔
29		4✔
30	conn.execute(	4✔
31	&stmt,	4✔
32	&[	4✔
33	&workflow_id,	4✔
34	&serde_json::to_value(&workflow).context(error::SerdeJson)?,	4✔
35	],
36	)
37	.await?;	4✔
38
39	Ok(workflow_id)	4✔
40	}	8✔
41
42	async fn load_workflow(&self, id: &WorkflowId) -> Result<Workflow> {	3✔
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?;	3✔
45	let stmt = conn	3✔
46	.prepare("SELECT workflow FROM workflows WHERE id = $1")	3✔
47	.await?;	3✔
48
49	let row = conn.query(&stmt, &[&id]).await?;	3✔
50
51	if row.is_empty() {	3✔
52	return Err(error::Error::NoWorkflowForGivenId);	×
53	}	3✔
54		3✔
55	Ok(serde_json::from_value(row[0].get(0)).context(error::SerdeJson)?)	3✔
56	}	6✔
57	}

geo-engine / geoengine / 5624216179

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