datasources/
create_enrichment_strategy.rs

1use crate::datasource_error::DatasourceError;
2use crate::enrichment::Enrichment;
3use crate::entity::Entity;
4use crate::repository::ICRUDRepository;
5use crate::result::Result;
6use crate::row::Row;
7use crate::schema::Schema;
8use crate::stored_enrichment::StoredEnrichment;
9use crate::value::Value;
10use async_trait::async_trait;
11use std::sync::Arc;
12
13#[async_trait]
14pub trait CreateEnrichmentStrategy<T: Entity = Row>: Send + Sync {
15    async fn create_children(
16        &self,
17        writer: Arc<dyn ICRUDRepository>,
18        enrichment: &Enrichment,
19        parent: &mut T,
20    ) -> Result<()>;
21
22    async fn create_linked(
23        &self,
24        writer: Arc<dyn ICRUDRepository>,
25        enrichment: &Enrichment,
26        parent: &mut T,
27    ) -> Result<()>;
28}
29
30fn missing(value: &Value) -> bool {
31    matches!(value, Value::Null | Value::Int(0))
32        || matches!(value, Value::Text(text) if text.is_empty())
33}
34
35pub(crate) fn children<T: Entity, TChild: Entity>(
36    enrichment: &Enrichment,
37    parent: &T,
38    schema: &Schema,
39) -> Result<Vec<TChild>> {
40    if enrichment.r#type == crate::EnrichmentPropertyType::Dictionary {
41        dictionary_children(enrichment, parent, schema)
42    } else {
43        Ok(parent.related(&enrichment.destination_property))
44    }
45}
46
47fn dictionary_children<T: Entity, TChild: Entity>(
48    enrichment: &Enrichment,
49    parent: &T,
50    schema: &Schema,
51) -> Result<Vec<TChild>> {
52    let key_name = enrichment.map_key()?;
53    parent
54        .payload(&enrichment.destination_property)
55        .into_iter()
56        .map(|row| {
57            let mut child = TChild::default();
58            let key = row
59                .get(key_name)
60                .cloned()
61                .or_else(|| row.get("key").cloned())
62                .unwrap_or(Value::Null);
63            set(&mut child, key_name, key)?;
64            if let Some(select) = enrichment.related_property.as_deref() {
65                let value = row.get(select).cloned().unwrap_or(Value::Null);
66                set(&mut child, select, value)?;
67            } else {
68                for name in enrichment.value_columns(schema)? {
69                    if let Some(cell) = row.get(&name).cloned() {
70                        set(&mut child, &name, cell)?;
71                    }
72                }
73            }
74            Ok(child)
75        })
76        .collect()
77}
78
79pub(crate) fn set<T: Entity>(item: &mut T, name: &str, value: Value) -> Result<()> {
80    item.set_cell(name, value)
81}
82
83pub(crate) async fn resolve_lookups<T: Entity>(
84    entries: &[StoredEnrichment<T>],
85    items: &mut [T],
86    record: &mut Row,
87    schema: &Schema,
88) -> Result<()> {
89    for item in items.iter_mut() {
90        for entry in entries {
91            let enrichment = &entry.enrichment;
92            let Some(related) = enrichment.related_property.as_deref() else {
93                continue;
94            };
95            if enrichment.association.is_some() {
96                resolve_linked(entry, item, related).await?;
97                continue;
98            }
99            if enrichment.r#type != crate::EnrichmentPropertyType::Object {
100                continue;
101            }
102            let dest = &enrichment.destination_property;
103            if record
104                .get(&enrichment.source_property)
105                .is_some_and(|value| !missing(value))
106            {
107                continue;
108            }
109            let Some(name) = record.get(dest).cloned().filter(|value| !missing(value)) else {
110                if schema
111                    .columns
112                    .get(dest)
113                    .is_some_and(|column| !column.nullable)
114                {
115                    return Err(DatasourceError::message(format!("Unknown {dest}")));
116                }
117                continue;
118            };
119            let found = entry
120                .lookup(related, name.clone(), &enrichment.related_key)
121                .await?;
122            let id = found
123                .filter(|value| !missing(value))
124                .ok_or_else(|| {
125                    DatasourceError::message(format!("Unknown {dest} \"{}\"", name.wire_text()))
126                })?;
127            set(record, &enrichment.source_property, id)?;
128        }
129    }
130    Ok(())
131}
132
133async fn resolve_linked<T: Entity>(
134    entry: &StoredEnrichment<T>,
135    item: &mut T,
136    related: &str,
137) -> Result<()> {
138    let enrichment = &entry.enrichment;
139    let mut rows = item.payload(&enrichment.destination_property);
140    let mut changed = false;
141    for row in &mut rows {
142        if row
143            .get(&enrichment.related_key)
144            .is_some_and(|value| !missing(value))
145        {
146            continue;
147        }
148        let Some(name) = row.get(related).cloned().filter(|value| !missing(value)) else {
149            continue;
150        };
151        let found = entry
152            .lookup(related, name.clone(), &enrichment.related_key)
153            .await?;
154        let id = found.filter(|value| !missing(value)).ok_or_else(|| {
155            DatasourceError::message(format!(
156                "Unknown {} \"{}\"",
157                enrichment.destination_property,
158                name.wire_text()
159            ))
160        })?;
161        row.insert(enrichment.related_key.clone(), id);
162        changed = true;
163    }
164    if changed {
165        item.set_payload(&enrichment.destination_property, rows)?;
166    }
167    Ok(())
168}
datasources/create_enrichment_strategy.rs

datasources/
create_enrichment_strategy.rs
