Ceres — Harvest-First Toolkit for Open Data Portals

Ceres is centered on harvesting and synchronizing open data metadata. Embeddings, semantic search, exports, and API access are downstream capabilities layered on top of the harvested catalog.

Repository: https://github.com/AndreaBozzo/Ceres License: Apache-2.0 | Rust edition: 2024 | MSRV: 1.88+

Pipeline

Metadata:  Portal URL → PortalClient (fetch) → DeltaDetector (content_hash) → DatasetStore (upsert, no embedding)
Embedding: DatasetStore (pending) → EmbeddingProvider (vector) → DatasetStore (update embedding)
Combined:  HarvestPipeline = HarvestService + EmbeddingService

Harvesting and embedding are decoupled: HarvestService handles metadata, EmbeddingService handles vectors, and HarvestPipeline composes both when you want a combined workflow. Metadata-only harvests require no embedding provider. Each stage is trait-based, so components can be swapped or mocked independently.

Current Product Shape

• Harvest first and keep metadata synchronized over time
• Add embeddings later only if you want semantic retrieval
• Prefer Ollama for local embedding, with Gemini and OpenAI still supported
• Support CKAN today and DCAT udata portals in the current client factory
• Expose search, export, and API workflows over the same harvested catalog

Crate Map

Core Traits (ceres-core::traits)

pub trait EmbeddingProvider: Send + Sync + Clone {
    fn name(&self) -> &'static str;
    fn dimension(&self) -> usize;
    fn generate(&self, text: &str) -> impl Future<Output = Result<Vec<f32>, AppError>> + Send;
    fn max_batch_size(&self) -> usize { 1 }
    fn generate_batch(&self, texts: &[String]) -> impl Future<Output = Result<Vec<Vec<f32>>, AppError>> + Send;
}

pub trait PortalClient: Send + Sync + Clone {
    type PortalData: Send;
    fn portal_type(&self) -> &'static str;
    fn base_url(&self) -> &str;
    fn list_dataset_ids(&self) -> impl Future<Output = Result<Vec<String>, AppError>> + Send;
    fn get_dataset(&self, id: &str) -> impl Future<Output = Result<Self::PortalData, AppError>> + Send;
    fn into_new_dataset(data: Self::PortalData, portal_url: &str, url_template: Option<&str>, language: &str) -> NewDataset;
    fn search_modified_since(&self, since: DateTime<Utc>) -> impl Future<Output = Result<Vec<Self::PortalData>, AppError>> + Send;
    fn search_all_datasets(&self) -> impl Future<Output = Result<Vec<Self::PortalData>, AppError>> + Send;
}

pub trait PortalClientFactory: Send + Sync + Clone {
    type Client: PortalClient;
    fn create(&self, portal_url: &str, portal_type: PortalType, language: &str) -> Result<Self::Client, AppError>;
}

pub trait DatasetStore: Send + Sync + Clone {
    fn get_by_id(&self, id: Uuid) -> impl Future<Output = Result<Option<Dataset>, AppError>> + Send;
    fn get_hashes_for_portal(&self, portal_url: &str) -> impl Future<Output = Result<HashMap<String, Option<String>>, AppError>> + Send;
    fn upsert(&self, dataset: &NewDataset) -> impl Future<Output = Result<Uuid, AppError>> + Send;
    fn batch_upsert(&self, datasets: &[NewDataset]) -> impl Future<Output = Result<Vec<Uuid>, AppError>> + Send;
    fn search(&self, query_vector: Vec<f32>, limit: usize) -> impl Future<Output = Result<Vec<SearchResult>, AppError>> + Send;
    fn list_stream<'a>(&'a self, portal_filter: Option<&'a str>, limit: Option<usize>) -> BoxStream<'a, Result<Dataset, AppError>>;
    fn get_last_sync_time(&self, portal_url: &str) -> impl Future<Output = Result<Option<DateTime<Utc>>, AppError>> + Send;
    fn record_sync_status(&self, portal_url: &str, sync_time: DateTime<Utc>, sync_mode: &str, sync_status: &str, datasets_synced: i32) -> impl Future<Output = Result<(), AppError>> + Send;
    fn health_check(&self) -> impl Future<Output = Result<(), AppError>> + Send;
    // + update_timestamp_only, batch_update_timestamps, get_duplicate_titles
    // Stale detection
    fn mark_stale_datasets(&self, portal_url: &str, sync_start: DateTime<Utc>) -> impl Future<Output = Result<u64, AppError>> + Send;
    fn mark_stale_by_exclusion(&self, portal_url: &str, seen_ids: &[String]) -> impl Future<Output = Result<u64, AppError>> + Send;
    // Pending embeddings
    fn list_pending_embeddings(&self, portal_filter: Option<&str>, limit: usize) -> impl Future<Output = Result<Vec<Dataset>, AppError>> + Send;
}

Key Types

Quick Start

# Install
cargo install ceres-search

# Start PostgreSQL + pgvector
docker compose up db -d

# Configure
cp .env.example .env

# Run migrations
make migrate

# Harvest metadata without embeddings
ceres harvest https://dati.comune.milano.it --metadata-only

# Or harvest a DCAT portal
ceres harvest https://data.public.lu --type dcat --metadata-only

# Optional: local embeddings through Ollama
export EMBEDDING_PROVIDER=ollama
ceres embed

# Search
ceres search "trasporto pubblico" --limit 5

# Export
ceres export --format jsonl > datasets.jsonl

# Stats
ceres stats

Reference Guides

Version Notes

• Current version: 0.3.5
• crates.io package: ceres-search
• Harvesting and embedding are decoupled: --metadata-only harvests without API key, embed command generates embeddings separately
• Ollama is the preferred local embedding path; Gemini and OpenAI remain available
• Current portal client factory supports CKAN and DCAT udata
• Stale dataset detection: datasets removed from portals are soft-marked (is_stale) during full syncs
• Supports Ollama, Gemini, and OpenAI embeddings
• 25+ pre-configured CKAN portals (354k+ datasets)
• HuggingFace dataset: AndreaBozzo/ceres-open-data-index