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

40

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

4✔

41

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

4✔

43

        let trans = conn

4✔

44

            .build_transaction()

4✔

45

            .start()

4✔

46

            .await

4✔

47

            .context(PostgresProjectDbError)?;

4✔

49

        let stmt = trans

4✔

50

            .prepare(&format!(

4✔

51

4✔

52

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

4✔

53

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

4✔

54

        WHERE

4✔

55

            u.user_id = $1

4✔

56

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

4✔

57

        ORDER BY p.{}

4✔

58

        LIMIT $2

4✔

59

        OFFSET $3;",

4✔

60

                options.order.to_sql_string()

4✔

61

))

4✔

62

            .await.context(PostgresProjectDbError)?;

4✔

64

        let project_rows = trans

4✔

65

            .query(

4✔

66

                &stmt,

4✔

67

&[

4✔

68

                    &self.session.user.id,

4✔

69

                    &i64::from(options.limit),

4✔

70

                    &i64::from(options.offset),

4✔

71

],

4✔

72

4✔

73

            .await

4✔

74

            .context(PostgresProjectDbError)?;

4✔

76

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

4✔

78

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

4✔

80

        Ok(project_listings)

4✔

81

8✔

83

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

50✔

84

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

50✔

86

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

50✔

88

        let trans = conn

50✔

89

            .build_transaction()

50✔

90

            .start()

50✔

91

            .await

50✔

92

            .context(PostgresProjectDbError)?;

50✔

94

        let version_id = insert_project(&trans, &project).await?;

50✔

96

        let stmt = trans

50✔

97

            .prepare(

50✔

98

                "INSERT INTO

50✔

99

                    project_version_authors (project_version_id, user_id)

50✔

100

                VALUES

50✔

101

                    ($1, $2);",

50✔

102

50✔

103

            .await

50✔

104

            .context(PostgresProjectDbError)?;

50✔

106

        trans

50✔

107

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

50✔

108

            .await

50✔

109

            .context(PostgresProjectDbError)?;

50✔

111

        let stmt = trans

50✔

112

            .prepare(

50✔

113

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

50✔

114

50✔

115

            .await

50✔

116

            .context(PostgresProjectDbError)?;

50✔

118

        trans

50✔

119

            .execute(

50✔

120

                &stmt,

50✔

121

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

50✔

122

50✔

123

            .await

50✔

124

            .context(PostgresProjectDbError)?;

50✔

126

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

50✔

128

        Ok(project.id)

50✔

129

100✔

131

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

34✔

132

        self.load_project_version(project, LoadVersion::Latest)

34✔

133

            .await

34✔

134

68✔

137

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

22✔

138

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

22✔

140

        let trans = conn

22✔

141

            .build_transaction()

22✔

142

            .start()

22✔

143

            .await

22✔

144

            .context(PostgresProjectDbError)?;

22✔

146

        self.ensure_permission_in_tx(update.id.into(), Permission::Owner, &trans)

22✔

147

            .await

22✔

148

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

22✔

150

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

22✔

152

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

22✔

154

        let stmt = trans

22✔

155

            .prepare(

22✔

156

                "INSERT INTO

22✔

157

                    project_version_authors (project_version_id, user_id)

22✔

158

                VALUES

22✔

159

                    ($1, $2);",

22✔

160

22✔

161

            .await

22✔

162

            .context(PostgresProjectDbError)?;

22✔

164

        trans

22✔

165

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

22✔

166

            .await

22✔

167

            .context(PostgresProjectDbError)?;

22✔

169

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

22✔

171

        Ok(())

22✔

172

44✔

174

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

4✔

175

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

4✔

176

        let trans = conn

4✔

177

            .build_transaction()

4✔

178

            .start()

4✔

179

            .await

4✔

180

            .context(PostgresProjectDbError)?;

4✔

182

        self.ensure_permission_in_tx(project.into(), Permission::Owner, &trans)

4✔

183

            .await

4✔

184

            .boxed_context(AccessFailedProjectDbError { project })?;

4✔

186

        let stmt = trans

3✔

187

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

3✔

188

            .await

3✔

189

            .context(PostgresProjectDbError)?;

3✔

191

        let rows_affected = trans

3✔

192

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

3✔

193

            .await

3✔

194

            .context(PostgresProjectDbError)?;

3✔

196

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

3✔

198

        ensure!(

3✔

199

            rows_affected == 1,

3✔

203

        Ok(())

3✔

204

8✔

211

    ) -> Result<Project, ProjectDbError> {

40✔

212

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

40✔

213

        let trans = conn

40✔

214

            .build_transaction()

40✔

215

            .start()

40✔

216

            .await

40✔

217

            .context(PostgresProjectDbError)?;

40✔

219

        self.ensure_permission_in_tx(project.into(), Permission::Owner, &trans)

40✔

220

            .await

40✔

221

            .boxed_context(AccessFailedProjectDbError { project })?;

40✔

223

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

37✔

224

            let stmt = trans

2✔

225

                .prepare(

2✔

226

2✔

227

            SELECT

2✔

228

                p.project_id,

2✔

229

                p.id,

2✔

230

                p.name,

2✔

231

                p.description,

2✔

232

                p.bounds,

2✔

233

                p.time_step,

2✔

234

                p.changed,

2✔

235

                a.user_id

2✔

236

            FROM

2✔

237

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

2✔

238

            WHERE p.project_id = $1 AND p.id = $2",

2✔

239

2✔

240

                .await

2✔

241

                .context(PostgresProjectDbError)?;

2✔

243

            let rows = trans

2✔

244

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

2✔

245

                .await

2✔

246

                .context(PostgresProjectDbError)?;

2✔

248

            if rows.is_empty() {

2✔

249

                return Err(ProjectDbError::ProjectVersionNotFound { project, version });

1✔

250

1✔

251

1✔

252

            rows

1✔

254

            let stmt = trans

35✔

255

                .prepare(

35✔

256

35✔

257

            SELECT

35✔

258

                p.project_id,

35✔

259

                p.id,

35✔

260

                p.name,

35✔

261

                p.description,

35✔

262

                p.bounds,

35✔

263

                p.time_step,

35✔

264

                p.changed,

35✔

265

                a.user_id

35✔

266

            FROM

35✔

267

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

35✔

268

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

35✔

269

                SELECT changed FROM project_versions WHERE project_id = $1

35✔

270

)",

35✔

271

35✔

272

                .await

35✔

273

                .context(PostgresProjectDbError)?;

35✔

275

            let rows = trans

35✔

276

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

35✔

277

                .await

35✔

278

                .context(PostgresProjectDbError)?;

35✔

280

            if rows.is_empty() {

35✔

282

35✔

283

35✔

284

            rows

35✔

287

        let row = &rows[0];

36✔

288

36✔

289

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

36✔

290

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

36✔

291

        let name = row.get(2);

36✔

292

        let description = row.get(3);

36✔

293

        let bounds = row.get(4);

36✔

294

        let time_step = row.get(5);

36✔

295

        let changed = row.get(6);

36✔

296

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

36✔

298

        let stmt = trans

36✔

299

            .prepare(

36✔

300

36✔

301

        SELECT

36✔

302

            name, workflow_id, symbology, visibility

36✔

303

        FROM project_version_layers

36✔

304

        WHERE project_version_id = $1

36✔

305

        ORDER BY layer_index ASC",

36✔

306

36✔

307

            .await

36✔

308

            .context(PostgresProjectDbError)?;

36✔

310

        let rows = trans

36✔

311

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

36✔

312

            .await

36✔

313

            .context(PostgresProjectDbError)?;

36✔

315

        let mut layers = vec![];

36✔

316

        for row in rows {

55✔

317

            layers.push(ProjectLayer {

19✔

318

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

19✔

319

                name: row.get(0),

19✔

320

                symbology: row.get(2),

19✔

321

                visibility: row.get(3),

19✔

322

});

19✔

323

19✔

325

        let project = Project {

36✔

326

            id: project_id,

36✔

327

            version: ProjectVersion {

36✔

328

                id: version_id,

36✔

329

                changed,

36✔

330

},

36✔

331

            name,

36✔

332

            description,

36✔

333

            layers,

36✔

334

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

36✔

335

            bounds,

36✔

336

            time_step,

36✔

337

};

36✔

338

36✔

339

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

36✔

341

        Ok(project)

36✔

342

80✔

347

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

8✔

348

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

8✔

349

        let trans = conn

8✔

350

            .build_transaction()

8✔

351

            .start()

8✔

352

            .await

8✔

353

            .context(PostgresProjectDbError)?;

8✔

355

        self.ensure_permission_in_tx(project.into(), Permission::Read, &trans)

8✔

356

            .await

8✔

357

            .boxed_context(AccessFailedProjectDbError { project })?;

8✔

359

        let stmt = trans

8✔

360

            .prepare(

8✔

361

8✔

362

                SELECT

8✔

363

                    p.id, p.changed, a.user_id

8✔

364

                FROM

8✔

365

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

8✔

366

                WHERE

8✔

367

                    project_id = $1

8✔

368

                ORDER BY

8✔

369

                    p.changed DESC, a.user_id DESC",

8✔

370

8✔

371

            .await.context(PostgresProjectDbError)?;

8✔

373

        let rows = trans

8✔

374

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

8✔

375

            .await

8✔

376

            .context(PostgresProjectDbError)?;

8✔

378

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

8✔

380

        Ok(rows

8✔

381

            .iter()

8✔

382

            .map(|row| ProjectVersion {

22✔

383

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

22✔

384

                changed: row.get(1),

22✔

385

})

22✔

386

            .collect())

8✔

387

16✔

390

pub async fn list_plots(

6✔

391

    trans: &Transaction<'_>,

6✔

392

    project_version_id: &ProjectVersionId,

6✔

393

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

6✔

394

    let stmt: tokio_postgres::Statement = trans

6✔

395

        .prepare(

6✔

396

6✔

397

                    SELECT name

6✔

398

                    FROM project_version_plots

6✔

399

                    WHERE project_version_id = $1;

6✔

400

",

6✔

401

6✔

402

        .await

6✔

403

        .context(PostgresProjectDbError)?;

6✔

405

    let plot_rows = trans

6✔

406

        .query(&stmt, &[project_version_id])

6✔

407

        .await

6✔

408

        .context(PostgresProjectDbError)?;

6✔

409

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

6✔

410

6✔

411

    Ok(plot_names)

6✔

412

6✔

414

pub async fn load_plots(

36✔

415

    trans: &Transaction<'_>,

36✔

416

    project_version_id: &ProjectVersionId,

36✔

417

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

36✔

418

    let stmt = trans

36✔

419

        .prepare(

36✔

420

36✔

421

                SELECT

36✔

422

                    name, workflow_id

36✔

423

                FROM project_version_plots

36✔

424

                WHERE project_version_id = $1

36✔

425

                ORDER BY plot_index ASC

36✔

426

",

36✔

427

36✔

428

        .await

36✔

429

        .context(PostgresProjectDbError)?;

36✔

431

    let rows = trans

36✔

432

        .query(&stmt, &[project_version_id])

36✔

433

        .await

36✔

434

        .context(PostgresProjectDbError)?;

36✔

436

    let plots = rows

36✔

437

        .into_iter()

36✔

438

        .map(|row| Plot {

36✔

439

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

14✔

440

            name: row.get(0),

14✔

441

})

36✔

442

        .collect();

36✔

443

36✔

444

    Ok(plots)

36✔

445

36✔

447

pub async fn update_plots(

22✔

448

    trans: &Transaction<'_>,

22✔

449

    project_id: &ProjectId,

22✔

450

    project_version_id: &ProjectVersionId,

22✔

451

    plots: &[Plot],

22✔

452

) -> Result<(), ProjectDbError> {

22✔

453

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

22✔

454

        let stmt = trans

10✔

455

            .prepare(

10✔

456

10✔

457

                    INSERT INTO project_version_plots (

10✔

458

                        project_id,

10✔

459

                        project_version_id,

10✔

460

                        plot_index,

10✔

461

                        name,

10✔

462

                        workflow_id)

10✔

463

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

10✔

464

",

10✔

465

10✔

466

            .await

10✔

467

            .context(PostgresProjectDbError)?;

10✔

469

        trans

10✔

470

            .execute(

10✔

471

                &stmt,

10✔

472

&[

10✔

473

                    project_id,

10✔

474

                    project_version_id,

10✔

475

                    &(idx as i32),

10✔

476

                    &plot.name,

10✔

477

                    &plot.workflow,

10✔

478

],

10✔

479

10✔

480

            .await

10✔

481

            .context(PostgresProjectDbError)?;

10✔

484

    Ok(())

22✔

485

22✔

487

pub async fn project_listings_from_rows(

4✔

488

    tx: &tokio_postgres::Transaction<'_>,

4✔

489

    project_rows: Vec<Row>,

4✔

490

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

4✔

491

    let mut project_listings = vec![];

4✔

492

    for project_row in project_rows {

10✔

493

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

6✔

494

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

6✔

495

        let name = project_row.get(2);

6✔

496

        let description = project_row.get(3);

6✔

497

        let changed = project_row.get(4);

6✔

499

        let stmt = tx

6✔

500

            .prepare(

6✔

501

6✔

502

                SELECT name

6✔

503

                FROM project_version_layers

6✔

504

                WHERE project_version_id = $1;",

6✔

505

6✔

506

            .await

6✔

507

            .context(PostgresProjectDbError)?;

6✔

509

        let layer_rows = tx

6✔

510

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

6✔

511

            .await

6✔

512

            .context(PostgresProjectDbError)?;

6✔

513

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

6✔

514

6✔

515

        project_listings.push(ProjectListing {

6✔

516

            id: project_id,

6✔

517

            name,

6✔

518

            description,

6✔

519

            layer_names,

6✔

520

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

6✔

521

            changed,

6✔

524

    Ok(project_listings)

4✔

525

4✔

527

pub async fn insert_project(

50✔

528

    trans: &Transaction<'_>,

50✔

529

    project: &Project,

50✔

530

) -> Result<ProjectVersionId, ProjectDbError> {

50✔

531

    let stmt = trans

50✔

532

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

50✔

533

        .await

50✔

534

        .context(PostgresProjectDbError)?;

50✔

536

    trans

50✔

537

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

50✔

538

        .await

50✔

539

        .context(PostgresProjectDbError)?;

50✔

541

    let stmt = trans

50✔

542

        .prepare(

50✔

543

            "INSERT INTO project_versions (

50✔

544

id,

50✔

545

                    project_id,

50✔

546

                    name,

50✔

547

                    description,

50✔

548

                    bounds,

50✔

549

                    time_step,

50✔

550

                    changed)

50✔

551

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

50✔

552

50✔

553

        .await

50✔

554

        .context(PostgresProjectDbError)?;

50✔

556

    let version_id = ProjectVersionId::new();

50✔

557

50✔

558

    trans

50✔

559

        .execute(

50✔

560

            &stmt,

50✔

561

&[

50✔

562

                &version_id,

50✔

563

                &project.id,

50✔

564

                &project.name,

50✔

565

                &project.description,

50✔

566

                &project.bounds,

50✔

567

                &project.time_step,

50✔

568

],

50✔

569

50✔

570

        .await

50✔

571

        .context(PostgresProjectDbError)?;

50✔

573

    Ok(version_id)

50✔

574

50✔

576

pub async fn update_project(

22✔

577

    trans: &Transaction<'_>,

22✔

578

    project: &Project,

22✔

579

    update: UpdateProject,

22✔

580

) -> Result<Project, ProjectDbError> {

22✔

581

    let project = project.update_project(update)?;

22✔

583

    let stmt = trans

22✔

584

        .prepare(

22✔

585

22✔

586

            INSERT INTO project_versions (

22✔

587

id,

22✔

588

                project_id,

22✔

589

                name,

22✔

590

                description,

22✔

591

                bounds,

22✔

592

                time_step,

22✔

593

                changed)

22✔

594

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

22✔

595

22✔

596

        .await

22✔

597

        .context(PostgresProjectDbError)?;

22✔

599

    trans

22✔

600

        .execute(

22✔

601

            &stmt,

22✔

602

&[

22✔

603

                &project.version.id,

22✔

604

                &project.id,

22✔

605

                &project.name,

22✔

606

                &project.description,

22✔

607

                &project.bounds,

22✔

608

                &project.time_step,

22✔

609

],

22✔

610

22✔

611

        .await

22✔

612

        .context(PostgresProjectDbError)?;

22✔

614

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

22✔

615

        let stmt = trans

18✔

616

            .prepare(

18✔

617

18✔

618

            INSERT INTO project_version_layers (

18✔

619

                project_id,

18✔

620

                project_version_id,

18✔

621

                layer_index,

18✔

622

                name,

18✔

623

                workflow_id,

18✔

624

                symbology,

18✔

625

                visibility)

18✔

626

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

18✔

627

18✔

628

            .await

18✔

629

            .context(PostgresProjectDbError)?;

18✔

631

        trans

18✔

632

            .execute(

18✔

633

                &stmt,

18✔

634

&[

18✔

635

                    &project.id,

18✔

636

                    &project.version.id,

18✔

637

                    &(idx as i32),

18✔

638

                    &layer.name,

18✔

639

                    &layer.workflow,

18✔

640

                    &layer.symbology,

18✔

641

                    &layer.visibility,

18✔

642

],

18✔

643

18✔

644

            .await

18✔

645

            .context(PostgresProjectDbError)?;

18✔

648

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

22✔

650

    Ok(project)

22✔

651

22✔

geo-engine / geoengine / 12910529774

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

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

geo-engine / geoengine / 12910529774

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