vantage_csv/
table_source.rs

1use crate::operation::CsvOperation;
2use async_trait::async_trait;
3use indexmap::IndexMap;
4use vantage_core::error;
5use vantage_dataset::traits::Result;
6use vantage_expressions::Expression;
7use vantage_expressions::Expressive;
8use vantage_expressions::traits::associated_expressions::AssociatedExpression;
9use vantage_expressions::traits::datasource::DataSource;
10use vantage_expressions::traits::expressive::{DeferredFn, ExpressiveEnum};
11use vantage_table::column::core::{Column, ColumnType};
12use vantage_table::table::Table;
13use vantage_table::traits::table_source::TableSource;
14use vantage_types::{Entity, Record};
15
16use crate::Csv;
17use crate::condition::apply_condition;
18use crate::type_system::AnyCsvType;
19
20impl DataSource for Csv {}
21
22#[async_trait]
23impl TableSource for Csv {
24    type Column<Type>
25        = Column<Type>
26    where
27        Type: ColumnType;
28    type AnyType = AnyCsvType;
29    type Value = AnyCsvType;
30    type Id = String;
31    type Condition = vantage_expressions::Expression<Self::Value>;
32    type Source = String;
33
34    fn eq_value_condition(&self, field: &str, value: Self::Value) -> Result<Self::Condition> {
35        let column: Column<AnyCsvType> = Column::new(field);
36        Ok(CsvOperation::eq(&column, value))
37    }
38
39    fn in_values_condition(
40        &self,
41        field: &str,
42        values: Vec<Self::Value>,
43    ) -> Result<Self::Condition> {
44        let column: Column<AnyCsvType> = Column::new(field);
45        let list = Expression::new("{}", vec![ExpressiveEnum::Scalar(AnyCsvType::new(values))]);
46        Ok(CsvOperation::in_(&column, list))
47    }
48
49    fn describe_condition(&self, condition: &Self::Condition) -> String {
50        condition.preview()
51    }
52
53    fn condition_column(&self, condition: &Self::Condition) -> Option<String> {
54        vantage_table::conditions::condition_column_of(condition)
55    }
56
57    fn create_column<Type: ColumnType>(&self, name: &str) -> Self::Column<Type> {
58        Column::new(name)
59    }
60
61    fn to_any_column<Type: ColumnType>(
62        &self,
63        column: Self::Column<Type>,
64    ) -> Self::Column<Self::AnyType> {
65        Column::from_column(column)
66    }
67
68    fn convert_any_column<Type: ColumnType>(
69        &self,
70        any_column: Self::Column<Self::AnyType>,
71    ) -> Option<Self::Column<Type>> {
72        Some(Column::from_column(any_column))
73    }
74
75    fn expr(
76        &self,
77        template: impl Into<String>,
78        parameters: Vec<ExpressiveEnum<Self::Value>>,
79    ) -> Expression<Self::Value> {
80        Expression::new(template, parameters)
81    }
82
83    fn search_table_condition<E>(
84        &self,
85        _table: &Table<Self, E>,
86        _search_value: &str,
87    ) -> Expression<Self::Value>
88    where
89        E: Entity<Self::Value>,
90    {
91        // Sentinel: `apply_condition` turns it into Unsupported instead of
92        // silently matching every row.
93        Expression::new(crate::operation::OP_SEARCH, vec![])
94    }
95
96    async fn list_table_values<E>(
97        &self,
98        table: &Table<Self, E>,
99    ) -> Result<IndexMap<Self::Id, Record<Self::Value>>>
100    where
101        E: Entity<Self::Value>,
102        Self: Sized,
103    {
104        let mut records = self.read_csv(table.table_name(), table.columns())?;
105
106        for condition in table.conditions() {
107            records = apply_condition(records, condition).await?;
108        }
109
110        Ok(records)
111    }
112
113    async fn get_table_value<E>(
114        &self,
115        table: &Table<Self, E>,
116        id: &Self::Id,
117    ) -> Result<Option<Record<Self::Value>>>
118    where
119        E: Entity<Self::Value>,
120        Self: Sized,
121    {
122        let records = self.read_csv(table.table_name(), table.columns())?;
123        Ok(records.get(id).cloned())
124    }
125
126    async fn get_table_some_value<E>(
127        &self,
128        table: &Table<Self, E>,
129    ) -> Result<Option<(Self::Id, Record<Self::Value>)>>
130    where
131        E: Entity<Self::Value>,
132        Self: Sized,
133    {
134        let records = self.read_csv(table.table_name(), table.columns())?;
135        Ok(records.into_iter().next())
136    }
137
138    async fn get_table_count<E>(&self, table: &Table<Self, E>) -> Result<i64>
139    where
140        E: Entity<Self::Value>,
141        Self: Sized,
142    {
143        let records = self.read_csv(table.table_name(), table.columns())?;
144        Ok(records.len() as i64)
145    }
146
147    async fn get_table_sum<E>(
148        &self,
149        _table: &Table<Self, E>,
150        _column: &Self::Column<Self::AnyType>,
151    ) -> Result<Self::Value>
152    where
153        E: Entity<Self::Value>,
154        Self: Sized,
155    {
156        Err(error!("Sum not implemented for CSV backend"))
157    }
158
159    async fn get_table_max<E>(
160        &self,
161        _table: &Table<Self, E>,
162        _column: &Self::Column<Self::AnyType>,
163    ) -> Result<Self::Value>
164    where
165        E: Entity<Self::Value>,
166        Self: Sized,
167    {
168        Err(error!("Max not implemented for CSV backend"))
169    }
170
171    async fn get_table_min<E>(
172        &self,
173        _table: &Table<Self, E>,
174        _column: &Self::Column<Self::AnyType>,
175    ) -> Result<Self::Value>
176    where
177        E: Entity<Self::Value>,
178        Self: Sized,
179    {
180        Err(error!("Min not implemented for CSV backend"))
181    }
182
183    async fn insert_table_value<E>(
184        &self,
185        _table: &Table<Self, E>,
186        _id: &Self::Id,
187        _record: &Record<Self::Value>,
188    ) -> Result<Record<Self::Value>>
189    where
190        E: Entity<Self::Value>,
191        Self: Sized,
192    {
193        Err(error!("CSV is a read-only data source"))
194    }
195
196    async fn replace_table_value<E>(
197        &self,
198        _table: &Table<Self, E>,
199        _id: &Self::Id,
200        _record: &Record<Self::Value>,
201    ) -> Result<Record<Self::Value>>
202    where
203        E: Entity<Self::Value>,
204        Self: Sized,
205    {
206        Err(error!("CSV is a read-only data source"))
207    }
208
209    async fn patch_table_value<E>(
210        &self,
211        _table: &Table<Self, E>,
212        _id: &Self::Id,
213        _partial: &Record<Self::Value>,
214    ) -> Result<Record<Self::Value>>
215    where
216        E: Entity<Self::Value>,
217        Self: Sized,
218    {
219        Err(error!("CSV is a read-only data source"))
220    }
221
222    async fn delete_table_value<E>(&self, _table: &Table<Self, E>, _id: &Self::Id) -> Result<()>
223    where
224        E: Entity<Self::Value>,
225        Self: Sized,
226    {
227        Err(error!("CSV is a read-only data source"))
228    }
229
230    async fn delete_table_all_values<E>(&self, _table: &Table<Self, E>) -> Result<()>
231    where
232        E: Entity<Self::Value>,
233        Self: Sized,
234    {
235        Err(error!("CSV is a read-only data source"))
236    }
237
238    async fn insert_table_return_id_value<E>(
239        &self,
240        _table: &Table<Self, E>,
241        _record: &Record<Self::Value>,
242    ) -> Result<Self::Id>
243    where
244        E: Entity<Self::Value>,
245        Self: Sized,
246    {
247        Err(error!("CSV is a read-only data source"))
248    }
249
250    fn related_in_condition<SourceE: Entity<Self::Value> + 'static>(
251        &self,
252        target_field: &str,
253        source_table: &Table<Self, SourceE>,
254        source_column: &str,
255    ) -> Self::Condition
256    where
257        Self: Sized,
258    {
259        let src_col = self.create_column::<Self::AnyType>(source_column);
260        let fk_values = self.column_table_values_expr(source_table, &src_col);
261        let tgt_col = self.create_column::<Self::AnyType>(target_field);
262        tgt_col.in_(fk_values.expr())
263    }
264
265    fn column_table_values_expr<'a, E, Type: ColumnType>(
266        &'a self,
267        table: &Table<Self, E>,
268        column: &Self::Column<Type>,
269    ) -> AssociatedExpression<'a, Self, Self::Value, Vec<Type>>
270    where
271        E: Entity<Self::Value> + 'static,
272        Self: Sized,
273    {
274        use vantage_expressions::{
275            expr_any,
276            traits::{associated_expressions::AssociatedExpression, datasource::ExprDataSource},
277        };
278
279        let table_clone = table.clone();
280        let col = column.name().to_string();
281        let csv = self.clone();
282
283        let inner = expr_any!("{}", {
284            DeferredFn::new(move || {
285                let csv = csv.clone();
286                let table = table_clone.clone();
287                let col = col.clone();
288                Box::pin(async move {
289                    let records = csv.list_table_values(&table).await?;
290                    let values: Vec<AnyCsvType> = records
291                        .values()
292                        .filter_map(|r| r.get(&col).cloned())
293                        .collect();
294                    Ok(ExpressiveEnum::Scalar(AnyCsvType::new(values)))
295                })
296            })
297        });
298
299        let expr = expr_any!("{}", { self.defer(inner) });
300        AssociatedExpression::new(expr, self)
301    }
302}
vantage_csv/table_source.rs

vantage_csv/
table_source.rs
