ecospgr

Shared loader utilities for the ecospg data pipeline. This package extracts duplicated R code from ecospg data loaders into a reusable library, providing database connection, catalog registration, partitioned table management, staging/ingest, S3 upload, and CLI argument parsing utilities.

Installation

install.packages("ecospgr", repos = "https://ldalby.r-universe.dev")

Usage

Loaders keep their unique download/business logic and delegate shared steps to ecospgr:

library(ecospgr)

pg <- epg_pg_connect(application_name = "my-loader")

epg_run_pipeline(
  con = pg, sf_obj = sf_obj,
  topic = "mytopic", dataset = "mydata", version_tag = "2025-01-01",
  source_meta = epg_build_source_meta(src_url),
  raw_url = raw_url,
  # Verify nothing was truncated before publishing (see below):
  expected_features = n_reported_by_source,
  # Loader-specific options:
  geom_type = "Geometry",
  geom_transform = "ST_Force2D(...)",
  delete_before_insert = TRUE,
  set_latest = FALSE,
  on_conflict_version = "reuse"
)

Do all downloading and reading before calling epg_run_pipeline(). Everything it touches in the database happens inside one dbWithTransaction(), so a failure leaves no registered version, no partition and no half-loaded data — but only if the network work is already done.

Publishing guards

vector.<dataset>_latest is what consumers read, so the pipeline refuses to publish a version that would quietly serve wrong or partial data.

Guard	Behaviour
`on_conflict_version = "reuse"` without `delete_before_insert = TRUE`	Error. Reusing a version id without clearing it appends a second full copy of every feature on a re-run, and nothing downstream can tell the copies apart.
`expected_features` short of rows inserted	Error. Upstream services cap responses silently (WFS at `count`, ArcGIS at `maxRecordCount`); pass the count the source reported and a truncated load cannot become `_latest`. Relax with `count_tolerance`.
Zero features	Error. An empty upstream response would otherwise blank out `_latest`. Override with `allow_empty = TRUE`.
Credentials in metadata	`epg_build_source_meta()` strips query strings and `epg_persist_source_meta()` redacts recursively. `meta` is readable by `warehouse_reader`, so a download URL carrying an API key must never land in `source_meta`.

Use epg_redact_url() on any URL that reaches a log or an error message — readLines(), sf::st_read() and curl all echo the full URL, including its query string, when they fail.

Exported functions

All exported functions use the epg_ prefix.

Area	Function	Description
Connection	`epg_pg_connect()`	Connect to PostgreSQL (reads `PG*` env vars)
Catalog	`epg_ensure_topic()`	Ensure a topic exists in the catalog
	`epg_ensure_dataset()`	Ensure a dataset exists in the catalog
	`epg_register_version()`	Register a new dataset version
	`epg_set_raw_file_url()`	Set the raw file URL for a version
	`epg_drop_version()`	Remove a bad version, its partition and catalog row
Partition	`epg_create_partition()`	Create a partitioned table (identifier-validated DDL)
Staging	`epg_write_staging()`	Write an sf object to a staging table
Ingest	`epg_insert_normalized()`	Insert from staging with configurable geom transform
Finalize	`epg_finalize_version()`	Set version as latest and drop staging table
Pipeline	`epg_run_pipeline()`	All-in-one pipeline within `dbWithTransaction`
S3	`epg_resolve_s3_config()`	Resolve S3 configuration from environment
	`epg_upload_raw_files()`	Upload raw files to S3
Source metadata	`epg_build_source_meta()`	Build source metadata list
	`epg_collect_source_meta_overrides()`	Collect source metadata overrides
	`epg_persist_source_meta()`	Persist source metadata to database
Args	`epg_parse_arg()`	Parse a CLI argument
	`epg_parse_json_payload()`	Parse a JSON payload from CLI
	`epg_parse_json_arg()`	Parse a JSON CLI argument
	`epg_read_json_file()`	Read and parse a JSON file
Utils	`epg_sanitize_name()`	Sanitize a string for use as an identifier
	`epg_detect_xsd_for_gml()`	Detect XSD schema file for a GML file
	`epg_redact_url()`	Mask credential query parameters in a URL
	`epg_url_without_query()`	Drop the query string from a URL
Validation	`epg_check_feature_count()`	Verify a load against the count the source reported

Development

# Run tests
Rscript -e 'devtools::test()'

# Full R CMD check
Rscript -e 'devtools::check()'

# Regenerate documentation
Rscript -e 'roxygen2::roxygenise()'

License

MIT