sbol_db_sqlite/repo/
cluster.rs

1//! The sequence cluster store over SQLite.
2//!
3//! Assignments live in `sbol_sequence_cluster`. A rebuild replaces the whole
4//! table in one transaction (`DELETE` then row-by-row insert), so readers only
5//! ever see a complete clustering.
6
7use async_trait::async_trait;
8use sbol_db_core::DomainError;
9use sbol_db_storage::{ClusterId, ClusterStore};
10use sqlx::Row;
11
12use crate::pool::db_err;
13use crate::SqlitePool;
14
15#[derive(Clone)]
16pub struct SqliteClusterStore {
17    pool: SqlitePool,
18}
19
20impl SqliteClusterStore {
21    pub fn new(pool: SqlitePool) -> Self {
22        Self { pool }
23    }
24}
25
26#[async_trait]
27impl ClusterStore for SqliteClusterStore {
28    async fn cluster_id_of(&self, iri: &str) -> Result<Option<ClusterId>, DomainError> {
29        let row =
30            sqlx::query("SELECT cluster_id FROM sbol_sequence_cluster WHERE sequence_iri = ?")
31                .bind(iri)
32                .fetch_optional(&self.pool)
33                .await
34                .map_err(db_err)?;
35        match row {
36            Some(r) => Ok(Some(ClusterId(
37                r.try_get::<i64, _>("cluster_id").map_err(db_err)?,
38            ))),
39            None => Ok(None),
40        }
41    }
42
43    async fn cluster_mates(&self, iri: &str) -> Result<Vec<String>, DomainError> {
44        let rows = sqlx::query(
45            r#"
46            SELECT mate.sequence_iri AS iri
47            FROM sbol_sequence_cluster self_row
48            JOIN sbol_sequence_cluster mate ON mate.cluster_id = self_row.cluster_id
49            WHERE self_row.sequence_iri = ?
50              AND mate.sequence_iri <> ?
51            "#,
52        )
53        .bind(iri)
54        .bind(iri)
55        .fetch_all(&self.pool)
56        .await
57        .map_err(db_err)?;
58        let mut out = Vec::with_capacity(rows.len());
59        for r in rows {
60            out.push(r.try_get::<String, _>("iri").map_err(db_err)?);
61        }
62        Ok(out)
63    }
64
65    async fn replace_clusters(&self, pairs: Vec<(String, ClusterId)>) -> Result<(), DomainError> {
66        let mut tx = self.pool.begin().await.map_err(db_err)?;
67        sqlx::query("DELETE FROM sbol_sequence_cluster")
68            .execute(&mut *tx)
69            .await
70            .map_err(db_err)?;
71        for (iri, cluster_id) in &pairs {
72            sqlx::query(
73                "INSERT INTO sbol_sequence_cluster (sequence_iri, cluster_id) VALUES (?, ?) \
74                 ON CONFLICT (sequence_iri) DO UPDATE SET cluster_id = excluded.cluster_id",
75            )
76            .bind(iri)
77            .bind(cluster_id.0)
78            .execute(&mut *tx)
79            .await
80            .map_err(db_err)?;
81        }
82        tx.commit().await.map_err(db_err)?;
83        Ok(())
84    }
85
86    async fn all_assignments(&self) -> Result<Vec<(String, ClusterId)>, DomainError> {
87        let rows = sqlx::query("SELECT sequence_iri, cluster_id FROM sbol_sequence_cluster")
88            .fetch_all(&self.pool)
89            .await
90            .map_err(db_err)?;
91        let mut out = Vec::with_capacity(rows.len());
92        for r in rows {
93            let iri = r.try_get::<String, _>("sequence_iri").map_err(db_err)?;
94            let cluster = ClusterId(r.try_get::<i64, _>("cluster_id").map_err(db_err)?);
95            out.push((iri, cluster));
96        }
97        Ok(out)
98    }
99
100    async fn assign_cluster(&self, iri: &str, cluster: ClusterId) -> Result<(), DomainError> {
101        sqlx::query(
102            "INSERT INTO sbol_sequence_cluster (sequence_iri, cluster_id) VALUES (?, ?) \
103             ON CONFLICT (sequence_iri) DO UPDATE SET cluster_id = excluded.cluster_id",
104        )
105        .bind(iri)
106        .bind(cluster.0)
107        .execute(&self.pool)
108        .await
109        .map_err(db_err)?;
110        Ok(())
111    }
112
113    async fn max_cluster_id(&self) -> Result<Option<ClusterId>, DomainError> {
114        let row = sqlx::query("SELECT MAX(cluster_id) AS max_id FROM sbol_sequence_cluster")
115            .fetch_one(&self.pool)
116            .await
117            .map_err(db_err)?;
118        Ok(row
119            .try_get::<Option<i64>, _>("max_id")
120            .map_err(db_err)?
121            .map(ClusterId))
122    }
123}
sbol_db_sqlite/repo/cluster.rs

sbol_db_sqlite/repo/
cluster.rs
