12013594587

Committed 25 Nov 2024 03:18PM UTC coverage: 90.45% (+0.005%) from 90.445%

Build # 12013594587

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #997 from geo-engine/add-debug-info-to-onnx-op

print debug log if onnx loading fails

Run Details

4 of 9 new or added lines in 1 file covered. (44.44%)

15 existing lines in 8 files now uncovered.

132882 of 146912 relevant lines covered (90.45%)

54795.39 hits per line

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

99.56

/services/src/projects/postgres_projectdb.rs

use crate::contexts::PostgresDb;
use crate::projects::error::ProjectNotFoundProjectDbError;
use crate::projects::Plot;
use crate::projects::ProjectLayer;
use crate::projects::{
    CreateProject, Project, ProjectDb, ProjectId, ProjectListOptions, ProjectListing,
    ProjectVersion, ProjectVersionId, UpdateProject,
};

use super::error::Bb8ProjectDbError;
use super::error::PostgresProjectDbError;
use super::error::ProjectDbError;
use super::LoadVersion;
use crate::util::Identifier;
use crate::workflows::workflow::WorkflowId;
use async_trait::async_trait;
use bb8_postgres::tokio_postgres::Transaction;
use bb8_postgres::{
    tokio_postgres::tls::MakeTlsConnect, tokio_postgres::tls::TlsConnect, tokio_postgres::Socket,
};
use snafu::ensure;
use snafu::ResultExt;
use tokio_postgres::Row;

pub async fn list_plots(
    trans: &Transaction<'_>,
    project_version_id: &ProjectVersionId,
) -> Result<Vec<String>, ProjectDbError> {
    let stmt: tokio_postgres::Statement = trans
        .prepare(
            "
                    SELECT name
                    FROM project_version_plots
                    WHERE project_version_id = $1;
                ",
        )
        .await
        .context(PostgresProjectDbError)?;

    let plot_rows = trans
        .query(&stmt, &[project_version_id])
        .await
        .context(PostgresProjectDbError)?;
    let plot_names = plot_rows.iter().map(|row| row.get(0)).collect();

    Ok(plot_names)
}

pub async fn load_plots(
    trans: &Transaction<'_>,
    project_version_id: &ProjectVersionId,
) -> Result<Vec<Plot>, ProjectDbError> {
    let stmt = trans
        .prepare(
            "
                SELECT  
                    name, workflow_id
                FROM project_version_plots
                WHERE project_version_id = $1
                ORDER BY plot_index ASC
                ",
        )
        .await
        .context(PostgresProjectDbError)?;

    let rows = trans
        .query(&stmt, &[project_version_id])
        .await
        .context(PostgresProjectDbError)?;

    let plots = rows
        .into_iter()
        .map(|row| Plot {
            workflow: WorkflowId(row.get(1)),
            name: row.get(0),
        })
        .collect();

    Ok(plots)
}

pub async fn update_plots(
    trans: &Transaction<'_>,
    project_id: &ProjectId,
    project_version_id: &ProjectVersionId,
    plots: &[Plot],
) -> Result<(), ProjectDbError> {
    for (idx, plot) in plots.iter().enumerate() {
        let stmt = trans
            .prepare(
                "
                    INSERT INTO project_version_plots (
                        project_id,
                        project_version_id,
                        plot_index,
                        name,
                        workflow_id)
                    VALUES ($1, $2, $3, $4, $5);
                    ",
            )
            .await
            .context(PostgresProjectDbError)?;

        trans
            .execute(
                &stmt,
                &[
                    project_id,
                    project_version_id,
                    &(idx as i32),
                    &plot.name,
                    &plot.workflow,
                ],
            )
            .await
            .context(PostgresProjectDbError)?;
    }

    Ok(())
}

pub async fn project_listings_from_rows(
    tx: &tokio_postgres::Transaction<'_>,
    project_rows: Vec<Row>,
) -> Result<Vec<ProjectListing>, ProjectDbError> {
    let mut project_listings = vec![];
    for project_row in project_rows {
        let project_version_id = ProjectVersionId(project_row.get(0));
        let project_id = ProjectId(project_row.get(1));
        let name = project_row.get(2);
        let description = project_row.get(3);
        let changed = project_row.get(4);

        let stmt = tx
            .prepare(
                "
                SELECT name
                FROM project_version_layers
                WHERE project_version_id = $1;",
            )
            .await
            .context(PostgresProjectDbError)?;

        let layer_rows = tx
            .query(&stmt, &[&project_version_id])
            .await
            .context(PostgresProjectDbError)?;
        let layer_names = layer_rows.iter().map(|row| row.get(0)).collect();

        project_listings.push(ProjectListing {
            id: project_id,
            name,
            description,
            layer_names,
            plot_names: list_plots(tx, &project_version_id).await?,
            changed,
        });
    }
    Ok(project_listings)
}

pub async fn insert_project(
    trans: &Transaction<'_>,
    project: &Project,
) -> Result<ProjectVersionId, ProjectDbError> {
    let stmt = trans
        .prepare("INSERT INTO projects (id) VALUES ($1);")
        .await
        .context(PostgresProjectDbError)?;

    trans
        .execute(&stmt, &[&project.id])
        .await
        .context(PostgresProjectDbError)?;

    let stmt = trans
        .prepare(
            "INSERT INTO project_versions (
                    id,
                    project_id,
                    name,
                    description,
                    bounds,
                    time_step,
                    changed)
                    VALUES ($1, $2, $3, $4, $5, $6, CURRENT_TIMESTAMP);",
        )
        .await
        .context(PostgresProjectDbError)?;

    let version_id = ProjectVersionId::new();

    trans
        .execute(
            &stmt,
            &[
                &version_id,
                &project.id,
                &project.name,
                &project.description,
                &project.bounds,
                &project.time_step,
            ],
        )
        .await
        .context(PostgresProjectDbError)?;

    Ok(version_id)
}

pub async fn update_project(
    trans: &Transaction<'_>,
    project: &Project,
    update: UpdateProject,
) -> Result<Project, ProjectDbError> {
    let project = project.update_project(update)?;

    let stmt = trans
        .prepare(
            "
            INSERT INTO project_versions (
                id,
                project_id,
                name,
                description,
                bounds,
                time_step,
                changed)
            VALUES ($1, $2, $3, $4, $5, $6, CURRENT_TIMESTAMP);",
        )
        .await
        .context(PostgresProjectDbError)?;

    trans
        .execute(
            &stmt,
            &[
                &project.version.id,
                &project.id,
                &project.name,
                &project.description,
                &project.bounds,
                &project.time_step,
            ],
        )
        .await
        .context(PostgresProjectDbError)?;

    for (idx, layer) in project.layers.iter().enumerate() {
        let stmt = trans
            .prepare(
                "
            INSERT INTO project_version_layers (
                project_id,
                project_version_id,
                layer_index,
                name,
                workflow_id,
                symbology,
                visibility)
            VALUES ($1, $2, $3, $4, $5, $6, $7);",
            )
            .await
            .context(PostgresProjectDbError)?;

        trans
            .execute(
                &stmt,
                &[
                    &project.id,
                    &project.version.id,
                    &(idx as i32),
                    &layer.name,
                    &layer.workflow,
                    &layer.symbology,
                    &layer.visibility,
                ],
            )
            .await
            .context(PostgresProjectDbError)?;
    }

    update_plots(trans, &project.id, &project.version.id, &project.plots).await?;

    Ok(project)
}

#[async_trait]
impl<Tls> ProjectDb 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 list_projects(
        &self,
        options: ProjectListOptions,
    ) -> Result<Vec<ProjectListing>, ProjectDbError> {
        // TODO: project filters

        let mut conn = self.conn_pool.get().await.context(Bb8ProjectDbError)?;
        let trans = conn
            .build_transaction()
            .start()
            .await
            .context(PostgresProjectDbError)?;

        let stmt = trans
            .prepare(&format!(
                "
        SELECT p.id, p.project_id, p.name, p.description, p.changed
        FROM project_versions p
        WHERE
            p.changed >= ALL (SELECT changed FROM project_versions WHERE project_id = p.project_id)
        ORDER BY p.{}
        LIMIT $1
        OFFSET $2;",
                options.order.to_sql_string()
            ))
            .await
            .context(PostgresProjectDbError)?;

        let project_rows = trans
            .query(
                &stmt,
                &[&i64::from(options.limit), &i64::from(options.offset)],
            )
            .await
            .context(PostgresProjectDbError)?;

        let project_listings = project_listings_from_rows(&trans, project_rows).await?;

        trans.commit().await.context(PostgresProjectDbError)?;

        Ok(project_listings)
    }

    async fn create_project(&self, create: CreateProject) -> Result<ProjectId, ProjectDbError> {
        let mut conn = self.conn_pool.get().await.context(Bb8ProjectDbError)?;

        let project: Project = Project::from_create_project(create);

        let trans = conn
            .build_transaction()
            .start()
            .await
            .context(PostgresProjectDbError)?;

        insert_project(&trans, &project).await?;

        trans.commit().await.context(PostgresProjectDbError)?;

        Ok(project.id)
    }

    async fn load_project(&self, project: ProjectId) -> Result<Project, ProjectDbError> {
        self.load_project_version(project, LoadVersion::Latest)
            .await
    }

    #[allow(clippy::too_many_lines)]
    async fn update_project(&self, update: UpdateProject) -> Result<(), ProjectDbError> {
        let mut conn = self.conn_pool.get().await.context(Bb8ProjectDbError)?;

        let trans = conn
            .build_transaction()
            .start()
            .await
            .context(PostgresProjectDbError)?;

        let project = self.load_project(update.id).await?; // TODO: move inside transaction?

        update_project(&trans, &project, update).await?;

        trans.commit().await.context(PostgresProjectDbError)?;

        Ok(())
    }

    async fn delete_project(&self, project: ProjectId) -> Result<(), ProjectDbError> {
        let conn = self.conn_pool.get().await.context(Bb8ProjectDbError)?;

        let stmt = conn
            .prepare("DELETE FROM projects WHERE id = $1;")
            .await
            .context(PostgresProjectDbError)?;

        let rows_affected = conn
            .execute(&stmt, &[&project])
            .await
            .context(PostgresProjectDbError)?;

        ensure!(
            rows_affected == 1,
            ProjectNotFoundProjectDbError { project }
        );

        Ok(())
    }

    #[allow(clippy::too_many_lines)]
    async fn load_project_version(
        &self,
        project: ProjectId,
        version: LoadVersion,
    ) -> Result<Project, ProjectDbError> {
        let mut conn = self.conn_pool.get().await.context(Bb8ProjectDbError)?;

        let trans = conn
            .build_transaction()
            .start()
            .await
            .context(PostgresProjectDbError)?;

        let rows = if let LoadVersion::Version(version) = version {
            let stmt = trans
                .prepare(
                    "
            SELECT 
                p.project_id, 
                p.id, 
                p.name, 
                p.description,
                p.bounds,
                p.time_step,
                p.changed
            FROM 
                project_versions p
            WHERE p.project_id = $1 AND p.id = $2",
                )
                .await
                .context(PostgresProjectDbError)?;

            let rows = trans
                .query(&stmt, &[&project, &version])
                .await
                .context(PostgresProjectDbError)?;

            if rows.is_empty() {
                return Err(ProjectDbError::ProjectVersionNotFound { project, version });
            }

            rows
        } else {
            let stmt = trans
                .prepare(
                    "
            SELECT  
                p.project_id, 
                p.id, 
                p.name, 
                p.description,
                p.bounds,
                p.time_step,
                p.changed
            FROM 
                project_versions p
            WHERE project_id = $1 AND p.changed >= ALL(
                SELECT changed FROM project_versions WHERE project_id = $1
            )",
                )
                .await
                .context(PostgresProjectDbError)?;

            let rows = trans
                .query(&stmt, &[&project])
                .await
                .context(PostgresProjectDbError)?;

            if rows.is_empty() {
                return Err(ProjectDbError::ProjectNotFound { project });
            }

            rows
        };

        let row = &rows[0];

        let project_id = ProjectId(row.get(0));
        let version_id = ProjectVersionId(row.get(1));
        let name = row.get(2);
        let description = row.get(3);
        let bounds = row.get(4);
        let time_step = row.get(5);
        let changed = row.get(6);

        let stmt = trans
            .prepare(
                "
        SELECT  
            name, workflow_id, symbology, visibility
        FROM project_version_layers
        WHERE project_version_id = $1
        ORDER BY layer_index ASC",
            )
            .await
            .context(PostgresProjectDbError)?;

        let rows = trans
            .query(&stmt, &[&version_id])
            .await
            .context(PostgresProjectDbError)?;

        let mut layers = vec![];
        for row in rows {
            layers.push(ProjectLayer {
                workflow: WorkflowId(row.get(1)),
                name: row.get(0),
                symbology: row.get(2),
                visibility: row.get(3),
            });
        }

        let project = Project {
            id: project_id,
            version: ProjectVersion {
                id: version_id,
                changed,
            },
            name,
            description,
            layers,
            plots: load_plots(&trans, &version_id).await?,
            bounds,
            time_step,
        };

        trans.commit().await.context(PostgresProjectDbError)?;

        Ok(project)
    }

    async fn list_project_versions(
        &self,
        project: ProjectId,
    ) -> Result<Vec<ProjectVersion>, ProjectDbError> {
        let conn = self.conn_pool.get().await.context(Bb8ProjectDbError)?;

        let stmt = conn
            .prepare(
                "
                SELECT 
                    id, changed
                FROM 
                    project_versions
                WHERE 
                    project_id = $1 
                ORDER BY 
                    changed DESC",
            )
            .await
            .context(PostgresProjectDbError)?;

        let rows = conn
            .query(&stmt, &[&project])
            .await
            .context(PostgresProjectDbError)?;

        Ok(rows
            .iter()
            .map(|row| ProjectVersion {
                id: ProjectVersionId(row.get(0)),
                changed: row.get(1),
            })
            .collect())
    }
}

1	use crate::contexts::PostgresDb;
2	use crate::projects::error::ProjectNotFoundProjectDbError;
3	use crate::projects::Plot;
4	use crate::projects::ProjectLayer;
5	use crate::projects::{
6	CreateProject, Project, ProjectDb, ProjectId, ProjectListOptions, ProjectListing,
7	ProjectVersion, ProjectVersionId, UpdateProject,
8	};
9
10	use super::error::Bb8ProjectDbError;
11	use super::error::PostgresProjectDbError;
12	use super::error::ProjectDbError;
13	use super::LoadVersion;
14	use crate::util::Identifier;
15	use crate::workflows::workflow::WorkflowId;
16	use async_trait::async_trait;
17	use bb8_postgres::tokio_postgres::Transaction;
18	use bb8_postgres::{
19	tokio_postgres::tls::MakeTlsConnect, tokio_postgres::tls::TlsConnect, tokio_postgres::Socket,
20	};
21	use snafu::ensure;
22	use snafu::ResultExt;
23	use tokio_postgres::Row;
24
25	pub async fn list_plots(	9✔
26	trans: &Transaction<'_>,	9✔
27	project_version_id: &ProjectVersionId,	9✔
28	) -> Result<Vec<String>, ProjectDbError> {	9✔
29	let stmt: tokio_postgres::Statement = trans	9✔
30	.prepare(	9✔
31	"	9✔
32	SELECT name	9✔
33	FROM project_version_plots	9✔
34	WHERE project_version_id = $1;	9✔
35	",	9✔
36	)	9✔
37	.await	8✔
38	.context(PostgresProjectDbError)?;	9✔
39
40	let plot_rows = trans	9✔
41	.query(&stmt, &[project_version_id])	9✔
42	.await	8✔
43	.context(PostgresProjectDbError)?;	9✔
44	let plot_names = plot_rows.iter().map(\|row\| row.get(0)).collect();	9✔
45		9✔
46	Ok(plot_names)	9✔
47	}	9✔
48
49	pub async fn load_plots(	45✔
50	trans: &Transaction<'_>,	45✔
51	project_version_id: &ProjectVersionId,	45✔
52	) -> Result<Vec<Plot>, ProjectDbError> {	45✔
53	let stmt = trans	45✔
54	.prepare(	45✔
55	"	45✔
56	SELECT	45✔
57	name, workflow_id	45✔
58	FROM project_version_plots	45✔
59	WHERE project_version_id = $1	45✔
60	ORDER BY plot_index ASC	45✔
61	",	45✔
62	)	45✔
63	.await	45✔
64	.context(PostgresProjectDbError)?;	45✔
65
66	let rows = trans	45✔
67	.query(&stmt, &[project_version_id])	45✔
68	.await	45✔
69	.context(PostgresProjectDbError)?;	45✔
70
71	let plots = rows	45✔
72	.into_iter()	45✔
73	.map(\|row\| Plot {	45✔
74	workflow: WorkflowId(row.get(1)),	17✔
75	name: row.get(0),	17✔
76	})	45✔
77	.collect();	45✔
78		45✔
79	Ok(plots)	45✔
80	}	45✔
81
82	pub async fn update_plots(	30✔
83	trans: &Transaction<'_>,	30✔
84	project_id: &ProjectId,	30✔
85	project_version_id: &ProjectVersionId,	30✔
86	plots: &[Plot],	30✔
87	) -> Result<(), ProjectDbError> {	30✔
88	for (idx, plot) in plots.iter().enumerate() {	30✔
89	let stmt = trans	13✔
90	.prepare(	13✔
91	"	13✔
92	INSERT INTO project_version_plots (	13✔
93	project_id,	13✔
94	project_version_id,	13✔
95	plot_index,	13✔
96	name,	13✔
97	workflow_id)	13✔
98	VALUES ($1, $2, $3, $4, $5);	13✔
99	",	13✔
100	)	13✔
101	.await	14✔
102	.context(PostgresProjectDbError)?;	13✔
103
104	trans	13✔
105	.execute(	13✔
106	&stmt,	13✔
107	&[	13✔
108	project_id,	13✔
109	project_version_id,	13✔
110	&(idx as i32),	13✔
111	&plot.name,	13✔
112	&plot.workflow,	13✔
113	],	13✔
114	)	13✔
115	.await	13✔
116	.context(PostgresProjectDbError)?;	13✔
117	}
118
119	Ok(())	30✔
120	}	30✔
121
122	pub async fn project_listings_from_rows(	6✔
123	tx: &tokio_postgres::Transaction<'_>,	6✔
124	project_rows: Vec<Row>,	6✔
125	) -> Result<Vec<ProjectListing>, ProjectDbError> {	6✔
126	let mut project_listings = vec![];	6✔
127	for project_row in project_rows {	15✔
128	let project_version_id = ProjectVersionId(project_row.get(0));	9✔
129	let project_id = ProjectId(project_row.get(1));	9✔
130	let name = project_row.get(2);	9✔
131	let description = project_row.get(3);	9✔
132	let changed = project_row.get(4);	9✔
133
134	let stmt = tx	9✔
135	.prepare(	9✔
136	"	9✔
137	SELECT name	9✔
138	FROM project_version_layers	9✔
139	WHERE project_version_id = $1;",	9✔
140	)	9✔
141	.await	8✔
142	.context(PostgresProjectDbError)?;	9✔
143
144	let layer_rows = tx	9✔
145	.query(&stmt, &[&project_version_id])	9✔
146	.await	9✔
147	.context(PostgresProjectDbError)?;	9✔
148	let layer_names = layer_rows.iter().map(\|row\| row.get(0)).collect();	9✔
149		9✔
150	project_listings.push(ProjectListing {	9✔
151	id: project_id,	9✔
152	name,	9✔
153	description,	9✔
154	layer_names,	9✔
155	plot_names: list_plots(tx, &project_version_id).await?,	16✔
156	changed,	9✔
157	});
158	}
159	Ok(project_listings)	6✔
160	}	6✔
161
162	pub async fn insert_project(	62✔
163	trans: &Transaction<'_>,	62✔
164	project: &Project,	62✔
165	) -> Result<ProjectVersionId, ProjectDbError> {	62✔
166	let stmt = trans	62✔
167	.prepare("INSERT INTO projects (id) VALUES ($1);")	62✔
168	.await	61✔
169	.context(PostgresProjectDbError)?;	62✔
170
171	trans	62✔
172	.execute(&stmt, &[&project.id])	62✔
173	.await	59✔
174	.context(PostgresProjectDbError)?;	62✔
175
176	let stmt = trans	62✔
177	.prepare(	62✔
178	"INSERT INTO project_versions (	62✔
179	id,	62✔
180	project_id,	62✔
181	name,	62✔
182	description,	62✔
183	bounds,	62✔
184	time_step,	62✔
185	changed)	62✔
186	VALUES ($1, $2, $3, $4, $5, $6, CURRENT_TIMESTAMP);",	62✔
187	)	62✔
188	.await	163✔
189	.context(PostgresProjectDbError)?;	62✔
190
191	let version_id = ProjectVersionId::new();	62✔
192		62✔
193	trans	62✔
194	.execute(	62✔
195	&stmt,	62✔
196	&[	62✔
197	&version_id,	62✔
198	&project.id,	62✔
199	&project.name,	62✔
200	&project.description,	62✔
201	&project.bounds,	62✔
202	&project.time_step,	62✔
203	],	62✔
204	)	62✔
205	.await	61✔
206	.context(PostgresProjectDbError)?;	62✔
207
208	Ok(version_id)	62✔
209	}	62✔
210
211	pub async fn update_project(	30✔
212	trans: &Transaction<'_>,	30✔
213	project: &Project,	30✔
214	update: UpdateProject,	30✔
215	) -> Result<Project, ProjectDbError> {	30✔
216	let project = project.update_project(update)?;	30✔
217
218	let stmt = trans	30✔
219	.prepare(	30✔
220	"	30✔
221	INSERT INTO project_versions (	30✔
222	id,	30✔
223	project_id,	30✔
224	name,	30✔
225	description,	30✔
226	bounds,	30✔
227	time_step,	30✔
228	changed)	30✔
229	VALUES ($1, $2, $3, $4, $5, $6, CURRENT_TIMESTAMP);",	30✔
230	)	30✔
231	.await	33✔
232	.context(PostgresProjectDbError)?;	30✔
233
234	trans	30✔
235	.execute(	30✔
236	&stmt,	30✔
237	&[	30✔
238	&project.version.id,	30✔
239	&project.id,	30✔
240	&project.name,	30✔
241	&project.description,	30✔
242	&project.bounds,	30✔
243	&project.time_step,	30✔
244	],	30✔
245	)	30✔
246	.await	33✔
247	.context(PostgresProjectDbError)?;	30✔
248
249	for (idx, layer) in project.layers.iter().enumerate() {	30✔
250	let stmt = trans	26✔
251	.prepare(	26✔
252	"	26✔
253	INSERT INTO project_version_layers (	26✔
254	project_id,	26✔
255	project_version_id,	26✔
256	layer_index,	26✔
257	name,	26✔
258	workflow_id,	26✔
259	symbology,	26✔
260	visibility)	26✔
261	VALUES ($1, $2, $3, $4, $5, $6, $7);",	26✔
262	)	26✔
263	.await	161✔
264	.context(PostgresProjectDbError)?;	26✔
265
266	trans	26✔
267	.execute(	26✔
268	&stmt,	26✔
269	&[	26✔
270	&project.id,	26✔
271	&project.version.id,	26✔
272	&(idx as i32),	26✔
273	&layer.name,	26✔
274	&layer.workflow,	26✔
275	&layer.symbology,	26✔
276	&layer.visibility,	26✔
277	],	26✔
278	)	26✔
279	.await	26✔
280	.context(PostgresProjectDbError)?;	26✔
281	}
282
283	update_plots(trans, &project.id, &project.version.id, &project.plots).await?;	30✔
284
285	Ok(project)	30✔
286	}	30✔
287
288	#[async_trait]
289	impl<Tls> ProjectDb for PostgresDb<Tls>
290	where
291	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static + std::fmt::Debug,
292	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
293	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
294	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
295	{
296	async fn list_projects(
297	&self,
298	options: ProjectListOptions,
299	) -> Result<Vec<ProjectListing>, ProjectDbError> {	3✔
300	// TODO: project filters
301
302	let mut conn = self.conn_pool.get().await.context(Bb8ProjectDbError)?;	3✔
303	let trans = conn	3✔
304	.build_transaction()	3✔
305	.start()	3✔
306	.await	2✔
307	.context(PostgresProjectDbError)?;	3✔
308
309	let stmt = trans	3✔
310	.prepare(&format!(	3✔
311	"	3✔
312	SELECT p.id, p.project_id, p.name, p.description, p.changed	3✔
313	FROM project_versions p	3✔
314	WHERE	3✔
315	p.changed >= ALL (SELECT changed FROM project_versions WHERE project_id = p.project_id)	3✔
316	ORDER BY p.{}	3✔
317	LIMIT $1	3✔
318	OFFSET $2;",	3✔
319	options.order.to_sql_string()	3✔
320	))	3✔
321	.await	2✔
322	.context(PostgresProjectDbError)?;	3✔
323
324	let project_rows = trans	3✔
325	.query(	3✔
326	&stmt,	3✔
327	&[&i64::from(options.limit), &i64::from(options.offset)],	3✔
328	)	3✔
329	.await	2✔
330	.context(PostgresProjectDbError)?;	3✔
331
332	let project_listings = project_listings_from_rows(&trans, project_rows).await?;	13✔
333
334	trans.commit().await.context(PostgresProjectDbError)?;	3✔
335
336	Ok(project_listings)	3✔
337	}	6✔
338
339	async fn create_project(&self, create: CreateProject) -> Result<ProjectId, ProjectDbError> {	40✔
340	let mut conn = self.conn_pool.get().await.context(Bb8ProjectDbError)?;	40✔
341
342	let project: Project = Project::from_create_project(create);	40✔
343
344	let trans = conn	40✔
345	.build_transaction()	40✔
346	.start()	40✔
347	.await	37✔
348	.context(PostgresProjectDbError)?;	40✔
349
350	insert_project(&trans, &project).await?;	227✔
351
352	trans.commit().await.context(PostgresProjectDbError)?;	40✔
353
354	Ok(project.id)	40✔
355	}	80✔
356
357	async fn load_project(&self, project: ProjectId) -> Result<Project, ProjectDbError> {	30✔
358	self.load_project_version(project, LoadVersion::Latest)	30✔
359	.await	837✔
360	}	60✔
361
362	#[allow(clippy::too_many_lines)]
363	async fn update_project(&self, update: UpdateProject) -> Result<(), ProjectDbError> {	19✔
364	let mut conn = self.conn_pool.get().await.context(Bb8ProjectDbError)?;	19✔
365
366	let trans = conn	19✔
367	.build_transaction()	19✔
368	.start()	19✔
369	.await	18✔
370	.context(PostgresProjectDbError)?;	19✔
371
372	let project = self.load_project(update.id).await?; // TODO: move inside transaction?	760✔
373
374	update_project(&trans, &project, update).await?;	215✔
375
376	trans.commit().await.context(PostgresProjectDbError)?;	21✔
377
378	Ok(())	19✔
379	}	38✔
380
381	async fn delete_project(&self, project: ProjectId) -> Result<(), ProjectDbError> {	3✔
382	let conn = self.conn_pool.get().await.context(Bb8ProjectDbError)?;	3✔
383
384	let stmt = conn	3✔
385	.prepare("DELETE FROM projects WHERE id = $1;")	3✔
386	.await	1✔
387	.context(PostgresProjectDbError)?;	3✔
388
389	let rows_affected = conn	3✔
390	.execute(&stmt, &[&project])	3✔
391	.await	2✔
392	.context(PostgresProjectDbError)?;	3✔
393
394	ensure!(	3✔
395	rows_affected == 1,	3✔
396	ProjectNotFoundProjectDbError { project }	1✔
397	);
398
399	Ok(())	2✔
400	}	6✔
401
402	#[allow(clippy::too_many_lines)]
403	async fn load_project_version(
404	&self,
405	project: ProjectId,
406	version: LoadVersion,
407	) -> Result<Project, ProjectDbError> {	35✔
408	let mut conn = self.conn_pool.get().await.context(Bb8ProjectDbError)?;	41✔
409
410	let trans = conn	35✔
411	.build_transaction()	35✔
412	.start()	35✔
413	.await	30✔
414	.context(PostgresProjectDbError)?;	35✔
415
416	let rows = if let LoadVersion::Version(version) = version {	35✔
417	let stmt = trans	2✔
418	.prepare(	2✔
419	"	2✔
420	SELECT	2✔
421	p.project_id,	2✔
422	p.id,	2✔
423	p.name,	2✔
424	p.description,	2✔
425	p.bounds,	2✔
426	p.time_step,	2✔
427	p.changed	2✔
428	FROM	2✔
429	project_versions p	2✔
430	WHERE p.project_id = $1 AND p.id = $2",	2✔
431	)	2✔
UNCOV 432	.await	×
433	.context(PostgresProjectDbError)?;	2✔
434
435	let rows = trans	2✔
436	.query(&stmt, &[&project, &version])	2✔
UNCOV 437	.await	×
438	.context(PostgresProjectDbError)?;	2✔
439
440	if rows.is_empty() {	2✔
441	return Err(ProjectDbError::ProjectVersionNotFound { project, version });	1✔
442	}	1✔
443		1✔
444	rows	1✔
445	} else {
446	let stmt = trans	33✔
447	.prepare(	33✔
448	"	33✔
449	SELECT	33✔
450	p.project_id,	33✔
451	p.id,	33✔
452	p.name,	33✔
453	p.description,	33✔
454	p.bounds,	33✔
455	p.time_step,	33✔
456	p.changed	33✔
457	FROM	33✔
458	project_versions p	33✔
459	WHERE project_id = $1 AND p.changed >= ALL(	33✔
460	SELECT changed FROM project_versions WHERE project_id = $1	33✔
461	)",	33✔
462	)	33✔
463	.await	237✔
464	.context(PostgresProjectDbError)?;	33✔
465
466	let rows = trans	33✔
467	.query(&stmt, &[&project])	33✔
468	.await	34✔
469	.context(PostgresProjectDbError)?;	33✔
470
471	if rows.is_empty() {	33✔
472	return Err(ProjectDbError::ProjectNotFound { project });	2✔
473	}	31✔
474		31✔
475	rows	31✔
476	};
477
478	let row = &rows[0];	32✔
479		32✔
480	let project_id = ProjectId(row.get(0));	32✔
481	let version_id = ProjectVersionId(row.get(1));	32✔
482	let name = row.get(2);	32✔
483	let description = row.get(3);	32✔
484	let bounds = row.get(4);	32✔
485	let time_step = row.get(5);	32✔
486	let changed = row.get(6);	32✔
487
488	let stmt = trans	32✔
489	.prepare(	32✔
490	"	32✔
491	SELECT	32✔
492	name, workflow_id, symbology, visibility	32✔
493	FROM project_version_layers	32✔
494	WHERE project_version_id = $1	32✔
495	ORDER BY layer_index ASC",	32✔
496	)	32✔
497	.await	455✔
498	.context(PostgresProjectDbError)?;	32✔
499
500	let rows = trans	32✔
501	.query(&stmt, &[&version_id])	32✔
502	.await	30✔
503	.context(PostgresProjectDbError)?;	32✔
504
505	let mut layers = vec![];	32✔
506	for row in rows {	49✔
507	layers.push(ProjectLayer {	17✔
508	workflow: WorkflowId(row.get(1)),	17✔
509	name: row.get(0),	17✔
510	symbology: row.get(2),	17✔
511	visibility: row.get(3),	17✔
512	});	17✔
513	}	17✔
514
515	let project = Project {	32✔
516	id: project_id,	32✔
517	version: ProjectVersion {	32✔
518	id: version_id,	32✔
519	changed,	32✔
520	},	32✔
521	name,	32✔
522	description,	32✔
523	layers,	32✔
524	plots: load_plots(&trans, &version_id).await?,	60✔
525	bounds,	32✔
526	time_step,	32✔
527	};	32✔
528		32✔
529	trans.commit().await.context(PostgresProjectDbError)?;	32✔
530
531	Ok(project)	32✔
532	}	70✔
533
534	async fn list_project_versions(
535	&self,
536	project: ProjectId,
537	) -> Result<Vec<ProjectVersion>, ProjectDbError> {	5✔
538	let conn = self.conn_pool.get().await.context(Bb8ProjectDbError)?;	5✔
539
540	let stmt = conn	5✔
541	.prepare(	5✔
542	"	5✔
543	SELECT	5✔
544	id, changed	5✔
545	FROM	5✔
546	project_versions	5✔
547	WHERE	5✔
548	project_id = $1	5✔
549	ORDER BY	5✔
550	changed DESC",	5✔
551	)	5✔
552	.await	5✔
553	.context(PostgresProjectDbError)?;	5✔
554
555	let rows = conn	5✔
556	.query(&stmt, &[&project])	5✔
557	.await	5✔
558	.context(PostgresProjectDbError)?;	5✔
559
560	Ok(rows	5✔
561	.iter()	5✔
562	.map(\|row\| ProjectVersion {	13✔
563	id: ProjectVersionId(row.get(0)),	13✔
564	changed: row.get(1),	13✔
565	})	13✔
566	.collect())	5✔
567	}	10✔
568	}

geo-engine / geoengine / 12013594587

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