qualia_core_db/domains/geospatial/adapters/
ckan_adapter.rs

1use crate::domains::geospatial::adapters::AdapterHttpRequest;
2use crate::net::disclosure::NetworkDisclosureRegistry;
3
4/// Adapter for dynamically discovering and fetching datasets from CKAN federated portals.
5pub struct CkanAdapter {
6    pub id: &'static str,
7    pub api_endpoint: String,
8}
9
10impl CkanAdapter {
11    pub fn new(id: &'static str, api_endpoint: &str) -> Self {
12        Self {
13            id,
14            api_endpoint: api_endpoint.to_string(),
15        }
16    }
17}
18
19impl super::DataAdapter for CkanAdapter {
20    fn adapter_id(&self) -> &'static str {
21        self.id
22    }
23
24    fn build_fetch_request(
25        &self,
26        bbox: (f64, f64, f64, f64),
27        _time_range: (u64, u64),
28        registry: &NetworkDisclosureRegistry,
29    ) -> Result<AdapterHttpRequest, String> {
30        let search_endpoint = format!("{}/action/package_search", self.api_endpoint);
31
32        if !registry.check_egress_consent(self.adapter_id(), &search_endpoint) {
33            return Err(format!(
34                "Consent denied or unregistered for CKAN endpoint {} by adapter {}",
35                search_endpoint,
36                self.adapter_id()
37            ));
38        }
39
40        let query = format!(
41            "{}?ext_bbox={},{},{},{}&rows=50",
42            search_endpoint, bbox.0, bbox.1, bbox.2, bbox.3
43        );
44
45        Ok(AdapterHttpRequest::get(query, "CKAN"))
46    }
47
48    fn needs_fetch_body(&self) -> bool {
49        true
50    }
51
52    fn handle_fetch_body(&self, body: &str) -> Result<(), String> {
53        let json: serde_json::Value = serde_json::from_str(body).map_err(|e| e.to_string())?;
54
55        let results = json
56            .get("result")
57            .and_then(|r| r.get("results"))
58            .and_then(|r| r.as_array())
59            .ok_or_else(|| "Failed to parse CKAN results array".to_string())?;
60
61        for dataset in results {
62            let title = dataset
63                .get("title")
64                .and_then(|v| v.as_str())
65                .unwrap_or("Untitled");
66            let license = dataset
67                .get("license_title")
68                .and_then(|v| v.as_str())
69                .unwrap_or("Unknown");
70            let created = dataset
71                .get("metadata_created")
72                .and_then(|v| v.as_str())
73                .unwrap_or("");
74
75            if let Some(resources) = dataset.get("resources").and_then(|v| v.as_array()) {
76                for res in resources {
77                    let format = res.get("format").and_then(|v| v.as_str()).unwrap_or("");
78                    let url = res.get("url").and_then(|v| v.as_str()).unwrap_or("");
79                    // In a full implementation, this would route to a dynamic ingestion queue.
80                    println!("Discovered CKAN Dataset '{}' | Format: {} | License: {} | Created: {} | URL: {}", 
81                             title, format, license, created, url);
82                }
83            }
84        }
85
86        Ok(())
87    }
88
89    fn primary_endpoint(&self) -> &str {
90        &self.api_endpoint
91    }
92
93    fn estimate_tile_count(&self, _bbox: (f64, f64, f64, f64)) -> u32 {
94        // We estimate 1 API query to search CKAN for datasets overlapping the bbox
95        1
96    }
97}
98
99#[cfg(test)]
100mod tests {
101    use super::*;
102    use crate::domains::geospatial::adapters::DataAdapter;
103
104    #[test]
105    fn test_ckan_adapter_egress() {
106        let adapter = CkanAdapter::new("ckan_test", "https://data.gov.au/data/api/3");
107        let mut registry = NetworkDisclosureRegistry::new();
108
109        let res1 = adapter.fetch_region((144.9, -37.9, 145.0, -37.8), (0, 0), &registry);
110        assert!(res1.is_err());
111
112        registry.register_egress(
113            adapter.adapter_id(),
114            "https://data.gov.au/data/api/3/action/package_search",
115            "Query federated CKAN repository for datasets",
116            "User executes spatial search",
117        );
118
119        let res2 = adapter.fetch_region((144.9, -37.9, 145.0, -37.8), (0, 0), &registry);
120        // During tests, we tolerate network timeouts/DNS issues
121        if let Err(e) = res2 {
122            assert!(
123                !e.contains("Consent denied"),
124                "Failed on consent when it should have been granted: {}",
125                e
126            );
127        }
128    }
129}
qualia_core_db/domains/geospatial/adapters/ckan_adapter.rs

qualia_core_db/domains/geospatial/adapters/
ckan_adapter.rs
