cratestack_sqlx/query/read/
projected_find_many.rs

1//! `find_many().select([...])` — projected multi-row read that
2//! returns `Vec<Projection<M>>`. Same partial-decode contract as
3//! [`super::projected_find_unique`].
4
5use cratestack_core::{CratestackContext, CratestackError};
6use cratestack_sql::{IntoColumnName, ReadSource};
7
8use crate::query::support::{ReadPolicyKind, push_order_and_paging, push_scoped_conditions};
9use crate::{FilterExpr, OrderClause, SqlxRuntime, sqlx};
10
11use super::find_many::FindMany;
12
13#[derive(Clone)]
14pub struct ProjectedFindMany<'a, M: 'static, PK: 'static> {
15    runtime: &'a SqlxRuntime,
16    descriptor: &'static dyn ReadSource<M, PK>,
17    filters: Vec<FilterExpr>,
18    order_by: Vec<OrderClause>,
19    limit: Option<i64>,
20    offset: Option<i64>,
21    for_update: bool,
22    selected: Vec<&'static str>,
23}
24
25impl<'a, M: 'static, PK: 'static> ProjectedFindMany<'a, M, PK> {
26    pub fn where_(mut self, filter: crate::Filter) -> Self {
27        self.filters.push(FilterExpr::from(filter));
28        self
29    }
30
31    pub fn where_expr(mut self, filter: FilterExpr) -> Self {
32        self.filters.push(filter);
33        self
34    }
35
36    pub fn where_any(mut self, filters: impl IntoIterator<Item = FilterExpr>) -> Self {
37        self.filters.push(FilterExpr::any(filters));
38        self
39    }
40
41    pub fn where_optional<F>(mut self, filter: Option<F>) -> Self
42    where
43        F: Into<FilterExpr>,
44    {
45        if let Some(filter) = filter {
46            self.filters.push(filter.into());
47        }
48        self
49    }
50
51    pub fn order_by(mut self, clause: OrderClause) -> Self {
52        self.order_by.push(clause);
53        self
54    }
55
56    pub fn limit(mut self, limit: i64) -> Self {
57        self.limit = Some(limit);
58        self
59    }
60
61    pub fn offset(mut self, offset: i64) -> Self {
62        self.offset = Some(offset);
63        self
64    }
65
66    pub fn for_update(mut self) -> Self {
67        self.for_update = true;
68        self
69    }
70
71    fn build_query(&self, ctx: &CratestackContext) -> sqlx::QueryBuilder<sqlx::Postgres> {
72        let mut query = sqlx::QueryBuilder::<sqlx::Postgres>::new("SELECT ");
73        query
74            .push(self.descriptor.select_projection_subset(&self.selected))
75            .push(" FROM ")
76            .push(self.descriptor.table_name());
77        push_scoped_conditions(
78            &mut query,
79            self.descriptor,
80            &self.filters,
81            None::<(&'static str, i64)>,
82            ctx,
83            ReadPolicyKind::List,
84        );
85        push_order_and_paging(&mut query, &self.order_by, self.limit, self.offset, ctx);
86        if self.for_update {
87            query.push(" FOR UPDATE");
88        }
89        query
90    }
91
92    pub async fn run(
93        self,
94        ctx: &CratestackContext,
95    ) -> Result<Vec<cratestack_sql::Projection<M>>, CratestackError>
96    where
97        M: crate::FromPartialPgRow,
98    {
99        // Inside an `@isolation` procedure: run on its transaction
100        // (docs/design/procedure-isolation.md §4).
101        if let Some(bound) = self.runtime.bound() {
102            return crate::bound::in_bound_savepoint!(bound, |sp| self.run_in_tx(sp, ctx));
103        }
104        let mut query = self.build_query(ctx);
105        let rows = query
106            .build()
107            .fetch_all(self.runtime.pool())
108            .await
109            .map_err(crate::error::cratestack_error_from_sqlx)?;
110        decode_many::<M>(rows, &self.selected)
111    }
112
113    pub async fn run_in_tx<'tx>(
114        self,
115        tx: &mut sqlx::Transaction<'tx, sqlx::Postgres>,
116        ctx: &CratestackContext,
117    ) -> Result<Vec<cratestack_sql::Projection<M>>, CratestackError>
118    where
119        M: crate::FromPartialPgRow,
120    {
121        let mut query = self.build_query(ctx);
122        let rows = query
123            .build()
124            .fetch_all(&mut **tx)
125            .await
126            .map_err(crate::error::cratestack_error_from_sqlx)?;
127        decode_many::<M>(rows, &self.selected)
128    }
129}
130
131fn decode_many<M>(
132    rows: Vec<sqlx::postgres::PgRow>,
133    selected: &[&'static str],
134) -> Result<Vec<cratestack_sql::Projection<M>>, CratestackError>
135where
136    M: crate::FromPartialPgRow,
137{
138    rows.into_iter()
139        .map(|row| {
140            M::decode_partial_pg_row(&row, selected)
141                .map(|value| cratestack_sql::Projection {
142                    value,
143                    selected: selected.to_vec(),
144                })
145                .map_err(crate::error::cratestack_error_from_sqlx)
146        })
147        .collect()
148}
149
150impl<'a, M: 'static, PK: 'static> FindMany<'a, M, PK> {
151    /// Restrict the SELECT to the named columns. See
152    /// [`super::find_unique::FindUnique::select`] for the caller-side
153    /// contract.
154    pub fn select<I, C>(self, columns: I) -> ProjectedFindMany<'a, M, PK>
155    where
156        I: IntoIterator<Item = C>,
157        C: IntoColumnName,
158    {
159        ProjectedFindMany {
160            runtime: self.runtime,
161            descriptor: self.descriptor,
162            filters: self.filters,
163            order_by: self.order_by,
164            limit: self.limit,
165            offset: self.offset,
166            for_update: self.for_update,
167            selected: columns
168                .into_iter()
169                .map(IntoColumnName::into_column_name)
170                .collect(),
171        }
172    }
173}
cratestack_sqlx/query/read/projected_find_many.rs

cratestack_sqlx/query/read/
projected_find_many.rs
