12466060820

Committed 23 Dec 2024 11:26AM UTC coverage: 90.353% (-0.2%) from 90.512%

Build # 12466060820

Build Type

Pull #998

github

Committed by

web-flow

Commit Message

Merge 66ab0655c into 34e12969f

Pull Request Pull Request #998: Quota and Data usage Logging

Run Details

834 of 1211 new or added lines in 66 files covered. (68.87%)

222 existing lines in 18 files now uncovered.

133834 of 148123 relevant lines covered (90.35%)

54353.18 hits per line

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

96.77

/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;

use super::registry::TxWorkflowRegistry;

#[async_trait]
impl<Tls> TxWorkflowRegistry for PostgresDb<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_in_tx(
        &self,
        workflow: Workflow,
        tx: &tokio_postgres::Transaction<'_>,
    ) -> Result<WorkflowId> {
        let workflow_id = WorkflowId::from_hash(&workflow);

        tx.execute(
            "INSERT INTO workflows (id, workflow) VALUES ($1, $2) 
            ON CONFLICT DO NOTHING;",
            &[
                &workflow_id,
                &serde_json::to_value(&workflow).context(error::SerdeJson)?,
            ],
        )
        .await?;

        Ok(workflow_id)
    }
}

#[async_trait]
impl<Tls> WorkflowRegistry for PostgresDb<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 mut conn = self.conn_pool.get().await?;
        let tx = conn.transaction().await?;

        let id = self.register_workflow_in_tx(workflow, &tx).await?;

        tx.commit().await?;

        Ok(id)
    }

    async fn load_workflow(&self, id: &WorkflowId) -> Result<Workflow> {
        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	use super::registry::TxWorkflowRegistry;
12
13	#[async_trait]
14	impl<Tls> TxWorkflowRegistry for PostgresDb<Tls>
15	where
16	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static + std::fmt::Debug,
17	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
18	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
19	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
20	{
21	async fn register_workflow_in_tx(
22	&self,
23	workflow: Workflow,
24	tx: &tokio_postgres::Transaction<'_>,
25	) -> Result<WorkflowId> {	131✔
26	let workflow_id = WorkflowId::from_hash(&workflow);	131✔
27		131✔
28	tx.execute(	131✔
29	"INSERT INTO workflows (id, workflow) VALUES ($1, $2)	131✔
30	ON CONFLICT DO NOTHING;",	131✔
31	&[	131✔
32	&workflow_id,	131✔
33	&serde_json::to_value(&workflow).context(error::SerdeJson)?,	131✔
34	],
35	)
36	.await?;	237✔
37
38	Ok(workflow_id)	131✔
39	}	262✔
40	}
41
42	#[async_trait]
43	impl<Tls> WorkflowRegistry for PostgresDb<Tls>
44	where
45	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static + std::fmt::Debug,
46	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
47	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
48	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
49	{
50	async fn register_workflow(&self, workflow: Workflow) -> Result<WorkflowId> {	109✔
51	let mut conn = self.conn_pool.get().await?;	109✔
52	let tx = conn.transaction().await?;	109✔
53
54	let id = self.register_workflow_in_tx(workflow, &tx).await?;	196✔
55
56	tx.commit().await?;	109✔
57
58	Ok(id)	109✔
59	}	218✔
60
61	async fn load_workflow(&self, id: &WorkflowId) -> Result<Workflow> {	33✔
62	let conn = self.conn_pool.get().await?;	33✔
63	let stmt = conn	33✔
64	.prepare("SELECT workflow FROM workflows WHERE id = $1")	33✔
65	.await?;	26✔
66
67	let row = conn.query(&stmt, &[&id]).await?;	33✔
68
69	if row.is_empty() {	33✔
UNCOV 70	return Err(error::Error::NoWorkflowForGivenId);	×
71	}	33✔
72		33✔
73	Ok(serde_json::from_value(row[0].get(0)).context(error::SerdeJson)?)	33✔
74	}	66✔
75	}

geo-engine / geoengine / 12466060820

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