sbol_db_sqlite/repo/
pagerank.rs

1//! The object PageRank store over SQLite.
2//!
3//! Scores live in `object_pagerank`. A rebuild replaces the whole table in one
4//! transaction (`DELETE` then row-by-row insert), so readers only ever see a
5//! complete ranking.
6
7use std::collections::HashMap;
8
9use async_trait::async_trait;
10use sbol_db_core::DomainError;
11use sbol_db_storage::{PageRankStore, RankRow};
12use sqlx::Row;
13
14use crate::pool::db_err;
15use crate::SqlitePool;
16
17#[derive(Clone)]
18pub struct SqlitePageRankStore {
19    pool: SqlitePool,
20}
21
22impl SqlitePageRankStore {
23    pub fn new(pool: SqlitePool) -> Self {
24        Self { pool }
25    }
26}
27
28#[async_trait]
29impl PageRankStore for SqlitePageRankStore {
30    async fn rank_of(&self, iri: &str) -> Result<Option<f64>, DomainError> {
31        let row = sqlx::query("SELECT score FROM object_pagerank WHERE iri = ?")
32            .bind(iri)
33            .fetch_optional(&self.pool)
34            .await
35            .map_err(db_err)?;
36        match row {
37            Some(r) => Ok(Some(r.try_get::<f64, _>("score").map_err(db_err)?)),
38            None => Ok(None),
39        }
40    }
41
42    async fn ranks_for(&self, iris: &[String]) -> Result<HashMap<String, f64>, DomainError> {
43        if iris.is_empty() {
44            return Ok(HashMap::new());
45        }
46        let placeholders = vec!["?"; iris.len()].join(", ");
47        let sql = format!("SELECT iri, score FROM object_pagerank WHERE iri IN ({placeholders})");
48        let mut q = sqlx::query(&sql);
49        for iri in iris {
50            q = q.bind(iri);
51        }
52        let rows = q.fetch_all(&self.pool).await.map_err(db_err)?;
53        let mut out = HashMap::with_capacity(rows.len());
54        for r in rows {
55            let iri: String = r.try_get("iri").map_err(db_err)?;
56            let score: f64 = r.try_get("score").map_err(db_err)?;
57            out.insert(iri, score);
58        }
59        Ok(out)
60    }
61
62    async fn all_ranks(&self) -> Result<Vec<RankRow>, DomainError> {
63        let rows = sqlx::query("SELECT iri, score FROM object_pagerank")
64            .fetch_all(&self.pool)
65            .await
66            .map_err(db_err)?;
67        let mut out = Vec::with_capacity(rows.len());
68        for r in rows {
69            let iri: String = r.try_get("iri").map_err(db_err)?;
70            let score: f64 = r.try_get("score").map_err(db_err)?;
71            out.push(RankRow { iri, score });
72        }
73        Ok(out)
74    }
75
76    async fn replace_all_ranks(&self, ranks: Vec<RankRow>) -> Result<(), DomainError> {
77        let mut tx = self.pool.begin().await.map_err(db_err)?;
78        sqlx::query("DELETE FROM object_pagerank")
79            .execute(&mut *tx)
80            .await
81            .map_err(db_err)?;
82        for row in &ranks {
83            sqlx::query(
84                "INSERT INTO object_pagerank (iri, score) VALUES (?, ?) \
85                 ON CONFLICT (iri) DO UPDATE SET score = excluded.score",
86            )
87            .bind(&row.iri)
88            .bind(row.score)
89            .execute(&mut *tx)
90            .await
91            .map_err(db_err)?;
92        }
93        tx.commit().await.map_err(db_err)?;
94        Ok(())
95    }
96}
sbol_db_sqlite/repo/pagerank.rs

sbol_db_sqlite/repo/
pagerank.rs
