geo-engine/geoengine | Build 6246101743 | services/src/pro/projects/postgres_projectdb.rs | Coveralls

28

async fn list_plots<Tls>(

4✔

29

    conn: &PooledConnection<'_, PostgresConnectionManager<Tls>>,

4✔

30

    project_version_id: &ProjectVersionId,

4✔

31

) -> Result<Vec<String>, ProjectDbError>

4✔

32

where

4✔

33

    Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,

4✔

34

    <Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,

4✔

35

    <Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,

4✔

36

    <<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,

4✔

37

4✔

38

    let stmt = conn

4✔

39

        .prepare(

4✔

40

4✔

41

                    SELECT name

4✔

42

                    FROM project_version_plots

4✔

43

                    WHERE project_version_id = $1;

4✔

44

",

4✔

45

4✔

46

        .await

4✔

47

        .context(PostgresProjectDbError)?;

4✔

49

    let plot_rows = conn

4✔

50

        .query(&stmt, &[project_version_id])

4✔

51

        .await

4✔

52

        .context(PostgresProjectDbError)?;

4✔

53

    let plot_names = plot_rows.iter().map(|row| row.get(0)).collect();

4✔

54

4✔

55

    Ok(plot_names)

4✔

56

4✔

58

async fn load_plots<Tls>(

13✔

59

    conn: &PooledConnection<'_, PostgresConnectionManager<Tls>>,

13✔

60

    project_version_id: &ProjectVersionId,

13✔

61

) -> Result<Vec<Plot>, ProjectDbError>

13✔

62

where

13✔

63

    Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,

13✔

64

    <Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,

13✔

65

    <Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,

13✔

66

    <<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,

13✔

67

13✔

68

    let stmt = conn

13✔

69

        .prepare(

13✔

70

13✔

71

                SELECT

13✔

72

                    name, workflow_id

13✔

73

                FROM project_version_plots

13✔

74

                WHERE project_version_id = $1

13✔

75

                ORDER BY plot_index ASC

13✔

76

",

13✔

77

13✔

78

        .await

14✔

79

        .context(PostgresProjectDbError)?;

13✔

81

    let rows = conn

13✔

82

        .query(&stmt, &[project_version_id])

13✔

83

        .await

14✔

84

        .context(PostgresProjectDbError)?;

13✔

86

    let plots = rows

13✔

87

        .into_iter()

13✔

88

        .map(|row| Plot {

13✔

89

            workflow: WorkflowId(row.get(1)),

6✔

90

            name: row.get(0),

6✔

91

})

13✔

92

        .collect();

13✔

93

13✔

94

    Ok(plots)

13✔

95

13✔

97

async fn update_plots(

11✔

98

    trans: &Transaction<'_>,

11✔

99

    project_id: &ProjectId,

11✔

100

    project_version_id: &ProjectVersionId,

11✔

101

    plots: &[Plot],

11✔

102

) -> Result<(), ProjectDbError> {

11✔

103

    for (idx, plot) in plots.iter().enumerate() {

11✔

104

        let stmt = trans

6✔

105

            .prepare(

6✔

106

6✔

107

                    INSERT INTO project_version_plots (

6✔

108

                        project_id,

6✔

109

                        project_version_id,

6✔

110

                        plot_index,

6✔

111

                        name,

6✔

112

                        workflow_id)

6✔

113

                    VALUES ($1, $2, $3, $4, $5);

6✔

114

",

6✔

115

6✔

116

            .await

6✔

117

            .context(PostgresProjectDbError)?;

6✔

119

        trans

6✔

120

            .execute(

6✔

121

                &stmt,

6✔

122

&[

6✔

123

                    project_id,

6✔

124

                    project_version_id,

6✔

125

                    &(idx as i32),

6✔

126

                    &plot.name,

6✔

127

                    &plot.workflow,

6✔

128

],

6✔

129

6✔

130

            .await

6✔

131

            .context(PostgresProjectDbError)?;

6✔

134

    Ok(())

11✔

135

11✔

145

    async fn list_projects(

2✔

146

        &self,

2✔

147

        options: ProjectListOptions,

2✔

148

    ) -> Result<Vec<ProjectListing>, ProjectDbError> {

2✔

149

        let conn = self.conn_pool.get().await.context(Bb8ProjectDbError)?;

2✔

151

        let stmt = conn

2✔

152

            .prepare(&format!(

2✔

153

2✔

154

        SELECT p.id, p.project_id, p.name, p.description, p.changed

2✔

155

        FROM user_permitted_projects u JOIN project_versions p ON (u.project_id = p.project_id)

2✔

156

        WHERE

2✔

157

            u.user_id = $1

2✔

158

            AND p.changed >= ALL (SELECT changed FROM project_versions WHERE project_id = p.project_id)

2✔

159

        ORDER BY p.{}

2✔

160

        LIMIT $2

2✔

161

        OFFSET $3;",

2✔

162

                options.order.to_sql_string()

2✔

163

))

2✔

164

            .await.context(PostgresProjectDbError)?;

2✔

166

        let project_rows = conn

2✔

167

            .query(

2✔

168

                &stmt,

2✔

169

&[

2✔

170

                    &self.session.user.id,

2✔

171

                    &i64::from(options.limit),

2✔

172

                    &i64::from(options.offset),

2✔

173

],

2✔

174

2✔

175

            .await

2✔

176

            .context(PostgresProjectDbError)?;

2✔

178

        let mut project_listings = vec![];

2✔

179

        for project_row in project_rows {

6✔

180

            let project_version_id = ProjectVersionId(project_row.get(0));

4✔

181

            let project_id = ProjectId(project_row.get(1));

4✔

182

            let name = project_row.get(2);

4✔

183

            let description = project_row.get(3);

4✔

184

            let changed = project_row.get(4);

4✔

186

            let stmt = conn

4✔

187

                .prepare(

4✔

188

4✔

189

                    SELECT name

4✔

190

                    FROM project_version_layers

4✔

191

                    WHERE project_version_id = $1;",

4✔

192

4✔

193

                .await

4✔

194

                .context(PostgresProjectDbError)?;

4✔

196

            let layer_rows = conn

4✔

197

                .query(&stmt, &[&project_version_id])

4✔

198

                .await

4✔

199

                .context(PostgresProjectDbError)?;

4✔

200

            let layer_names = layer_rows.iter().map(|row| row.get(0)).collect();

4✔

201

4✔

202

            project_listings.push(ProjectListing {

4✔

203

                id: project_id,

4✔

204

                name,

4✔

205

                description,

4✔

206

                layer_names,

4✔

207

                plot_names: list_plots(&conn, &project_version_id).await?,

8✔

208

                changed,

4✔

211

        Ok(project_listings)

2✔

212

4✔

214

    async fn create_project(&self, create: CreateProject) -> Result<ProjectId, ProjectDbError> {

22✔

215

        let mut conn = self.conn_pool.get().await.context(Bb8ProjectDbError)?;

22✔

217

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

22✔

219

        let trans = conn

22✔

220

            .build_transaction()

22✔

221

            .start()

22✔

222

            .await

22✔

223

            .context(PostgresProjectDbError)?;

22✔

225

        let stmt = trans

22✔

226

            .prepare("INSERT INTO projects (id) VALUES ($1);")

22✔

227

            .await

22✔

228

            .context(PostgresProjectDbError)?;

22✔

230

        trans

22✔

231

            .execute(&stmt, &[&project.id])

22✔

232

            .await

22✔

233

            .context(PostgresProjectDbError)?;

22✔

235

        let stmt = trans

22✔

236

            .prepare(

22✔

237

                "INSERT INTO project_versions (

22✔

238

id,

22✔

239

                    project_id,

22✔

240

                    name,

22✔

241

                    description,

22✔

242

                    bounds,

22✔

243

                    time_step,

22✔

244

                    changed)

22✔

245

                    VALUES ($1, $2, $3, $4, $5, $6, CURRENT_TIMESTAMP);",

22✔

246

22✔

247

            .await

44✔

248

            .context(PostgresProjectDbError)?;

22✔

250

        let version_id = ProjectVersionId::new();

22✔

251

22✔

252

        trans

22✔

253

            .execute(

22✔

254

                &stmt,

22✔

255

&[

22✔

256

                    &version_id,

22✔

257

                    &project.id,

22✔

258

                    &project.name,

22✔

259

                    &project.description,

22✔

260

                    &project.bounds,

22✔

261

                    &project.time_step,

22✔

262

],

22✔

263

22✔

264

            .await

21✔

265

            .context(PostgresProjectDbError)?;

22✔

267

        let stmt = trans

22✔

268

            .prepare(

22✔

269

                "INSERT INTO

22✔

270

                    project_version_authors (project_version_id, user_id)

22✔

271

                VALUES

22✔

272

                    ($1, $2);",

22✔

273

22✔

274

            .await

22✔

275

            .context(PostgresProjectDbError)?;

22✔

277

        trans

22✔

278

            .execute(&stmt, &[&version_id, &self.session.user.id])

22✔

279

            .await

21✔

280

            .context(PostgresProjectDbError)?;

22✔

282

        let stmt = trans

22✔

283

            .prepare(

22✔

284

                "INSERT INTO permissions (role_id, permission, project_id) VALUES ($1, $2, $3);",

22✔

285

22✔

286

            .await

28✔

287

            .context(PostgresProjectDbError)?;

22✔

289

        trans

22✔

290

            .execute(

22✔

291

                &stmt,

22✔

292

                &[&self.session.user.id, &Permission::Owner, &project.id],

22✔

293

22✔

294

            .await

21✔

295

            .context(PostgresProjectDbError)?;

22✔

297

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

22✔

299

        Ok(project.id)

22✔

300

44✔

302

    async fn load_project(&self, project: ProjectId) -> Result<Project, ProjectDbError> {

13✔

303

        self.load_project_version(project, LoadVersion::Latest)

13✔

304

            .await

352✔

305

26✔

308

    async fn update_project(&self, update: UpdateProject) -> Result<(), ProjectDbError> {

11✔

309

        let mut conn = self.conn_pool.get().await.context(Bb8ProjectDbError)?;

12✔

311

        self.ensure_permission(update.id, Permission::Owner)

11✔

312

            .await

54✔

313

            .boxed_context(AccessFailedProjectDbError { project: update.id })?;

11✔

315

        let trans = conn

11✔

316

            .build_transaction()

11✔

317

            .start()

11✔

318

            .await

12✔

319

            .context(PostgresProjectDbError)?;

11✔

321

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

344✔

323

        let project = project.update_project(update)?;

11✔

325

        let stmt = trans

11✔

326

            .prepare(

11✔

327

11✔

328

                INSERT INTO project_versions (

11✔

329

id,

11✔

330

                    project_id,

11✔

331

                    name,

11✔

332

                    description,

11✔

333

                    bounds,

11✔

334

                    time_step,

11✔

335

                    changed)

11✔

336

                VALUES ($1, $2, $3, $4, $5, $6, CURRENT_TIMESTAMP);",

11✔

337

11✔

338

            .await

63✔

339

            .context(PostgresProjectDbError)?;

11✔

341

        trans

11✔

342

            .execute(

11✔

343

                &stmt,

11✔

344

&[

11✔

345

                    &project.version.id,

11✔

346

                    &project.id,

11✔

347

                    &project.name,

11✔

348

                    &project.description,

11✔

349

                    &project.bounds,

11✔

350

                    &project.time_step,

11✔

351

],

11✔

352

11✔

353

            .await

12✔

354

            .context(PostgresProjectDbError)?;

11✔

356

        let stmt = trans

11✔

357

            .prepare(

11✔

358

                "INSERT INTO

11✔

359

                    project_version_authors (project_version_id, user_id)

11✔

360

                VALUES

11✔

361

                    ($1, $2);",

11✔

362

11✔

363

            .await

12✔

364

            .context(PostgresProjectDbError)?;

11✔

366

        trans

11✔

367

            .execute(&stmt, &[&project.version.id, &self.session.user.id])

11✔

368

            .await

11✔

369

            .context(PostgresProjectDbError)?;

11✔

371

        for (idx, layer) in project.layers.iter().enumerate() {

11✔

372

            let stmt = trans

11✔

373

                .prepare(

11✔

374

11✔

375

                INSERT INTO project_version_layers (

11✔

376

                    project_id,

11✔

377

                    project_version_id,

11✔

378

                    layer_index,

11✔

379

                    name,

11✔

380

                    workflow_id,

11✔

381

                    symbology,

11✔

382

                    visibility)

11✔

383

                VALUES ($1, $2, $3, $4, $5, $6, $7);",

11✔

384

11✔

385

                .await

104✔

386

                .context(PostgresProjectDbError)?;

11✔

388

            trans

11✔

389

                .execute(

11✔

390

                    &stmt,

11✔

391

&[

11✔

392

                        &project.id,

11✔

393

                        &project.version.id,

11✔

394

                        &(idx as i32),

11✔

395

                        &layer.name,

11✔

396

                        &layer.workflow,

11✔

397

                        &layer.symbology,

11✔

398

                        &layer.visibility,

11✔

399

],

11✔

400

11✔

401

                .await

12✔

402

                .context(PostgresProjectDbError)?;

11✔

405

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

12✔

407

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

12✔

409

        Ok(())

11✔

410

22✔

412

    async fn delete_project(&self, project: ProjectId) -> Result<(), ProjectDbError> {

2✔

413

        let conn = self.conn_pool.get().await.context(Bb8ProjectDbError)?;

3✔

415

        self.ensure_permission(project, Permission::Owner)

2✔

416

            .await

6✔

417

            .boxed_context(AccessFailedProjectDbError { project })?;

2✔

419

        let stmt = conn

2✔

420

            .prepare("DELETE FROM projects WHERE id = $1;")

2✔

421

            .await

2✔

422

            .context(PostgresProjectDbError)?;

2✔

424

        let rows_affected = conn

2✔

425

            .execute(&stmt, &[&project])

2✔

426

            .await

2✔

427

            .context(PostgresProjectDbError)?;

2✔

429

        ensure!(

2✔

430

            rows_affected == 1,

2✔

434

        Ok(())

2✔

435

4✔

438

    async fn load_project_version(

15✔

439

        &self,

15✔

440

        project: ProjectId,

15✔

441

        version: LoadVersion,

15✔

442

    ) -> Result<Project, ProjectDbError> {

15✔

443

        let conn = self.conn_pool.get().await.context(Bb8ProjectDbError)?;

16✔

445

        self.ensure_permission(project, Permission::Owner)

15✔

446

            .await

85✔

447

            .boxed_context(AccessFailedProjectDbError { project })?;

15✔

449

        let rows = if let LoadVersion::Version(version) = version {

13✔

480

            let stmt = conn

13✔

481

                .prepare(

13✔

482

13✔

483

            SELECT

13✔

484

                p.project_id,

13✔

485

                p.id,

13✔

486

                p.name,

13✔

487

                p.description,

13✔

488

                p.bounds,

13✔

489

                p.time_step,

13✔

490

                p.changed,

13✔

491

                a.user_id

13✔

492

            FROM

13✔

493

                project_versions p JOIN project_version_authors a ON (p.id = a.project_version_id)

13✔

494

            WHERE project_id = $1 AND p.changed >= ALL(

13✔

495

                SELECT changed FROM project_versions WHERE project_id = $1

13✔

496

)",

13✔

497

13✔

498

                .await

91✔

499

                .context(PostgresProjectDbError)?;

13✔

501

            let rows = conn

13✔

502

                .query(&stmt, &[&project])

13✔

503

                .await

13✔

504

                .context(PostgresProjectDbError)?;

13✔

506

            if rows.is_empty() {

13✔

508

13✔

509

13✔

510

            rows

13✔

513

        let row = &rows[0];

13✔

514

13✔

515

        let project_id = ProjectId(row.get(0));

13✔

516

        let version_id = ProjectVersionId(row.get(1));

13✔

517

        let name = row.get(2);

13✔

518

        let description = row.get(3);

13✔

519

        let bounds = row.get(4);

13✔

520

        let time_step = row.get(5);

13✔

521

        let changed = row.get(6);

13✔

522

        let _author_id = UserId(row.get(7));

13✔

524

        let stmt = conn

13✔

525

            .prepare(

13✔

526

13✔

527

        SELECT

13✔

528

            name, workflow_id, symbology, visibility

13✔

529

        FROM project_version_layers

13✔

530

        WHERE project_version_id = $1

13✔

531

        ORDER BY layer_index ASC",

13✔

532

13✔

533

            .await

194✔

534

            .context(PostgresProjectDbError)?;

13✔

536

        let rows = conn

13✔

537

            .query(&stmt, &[&version_id])

13✔

538

            .await

14✔

539

            .context(PostgresProjectDbError)?;

13✔

541

        let mut layers = vec![];

13✔

542

        for row in rows {

21✔

543

            layers.push(ProjectLayer {

8✔

544

                workflow: WorkflowId(row.get(1)),

8✔

545

                name: row.get(0),

8✔

546

                symbology: row.get(2),

8✔

547

                visibility: row.get(3),

8✔

548

});

8✔

549

8✔

552

            id: project_id,

13✔

553

            version: ProjectVersion {

13✔

554

                id: version_id,

13✔

555

                changed,

13✔

556

},

13✔

557

            name,

13✔

558

            description,

13✔

559

            layers,

13✔

560

            plots: load_plots(&conn, &version_id).await?,

28✔

561

            bounds,

13✔

562

            time_step,

13✔

564

30✔

566

    async fn list_project_versions(

6✔

567

        &self,

6✔

568

        project: ProjectId,

6✔

569

    ) -> Result<Vec<ProjectVersion>, ProjectDbError> {

6✔

570

        let conn = self.conn_pool.get().await.context(Bb8ProjectDbError)?;

6✔

572

        self.ensure_permission(project, Permission::Read)

6✔

573

            .await

20✔

574

            .boxed_context(AccessFailedProjectDbError { project })?;

6✔

576

        let stmt = conn

6✔

577

            .prepare(

6✔

578

6✔

579

                SELECT

6✔

580

                    p.id, p.changed, a.user_id

6✔

581

                FROM

6✔

582

                    project_versions p JOIN project_version_authors a ON (p.id = a.project_version_id)

6✔

583

                WHERE

6✔

584

                    project_id = $1

6✔

585

                ORDER BY

6✔

586

                    p.changed DESC, a.user_id DESC",

6✔

587

6✔

588

            .await.context(PostgresProjectDbError)?;

6✔

590

        let rows = conn

6✔

591

            .query(&stmt, &[&project])

6✔

592

            .await

6✔

593

            .context(PostgresProjectDbError)?;

6✔

595

        Ok(rows

6✔

596

            .iter()

6✔

597

            .map(|row| ProjectVersion {

18✔

598

                id: ProjectVersionId(row.get(0)),

18✔

599

                changed: row.get(1),

18✔

600

})

18✔

601

            .collect())

6✔

602

12✔

geo-engine / geoengine / 6246101743

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

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

geo-engine / geoengine / 6246101743

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