src/pxweb.rs
//! **#2 PxWeb / JSON-stat2 client** — one connector covers SCB, Energimyndigheten
//! and most Swedish statistics agencies (thousands of CC0 tables). It lists a
//! PxWeb table tree, GET/POSTs a cell selection, and **normalises the JSON-stat2
//! cube → Arrow** → Iceberg. Same client, different `base_url` per agency.
//!
//! The cube shaper [`parse_jsonstat2`] is the offline contract: a canned
//! JSON-stat2 cube → an Arrow table with one code column per dimension (in
//! `id` order) plus a nullable `value` column, cells decoded row-major (last
//! dimension varies fastest — the JSON-stat ordering).
//!
//! # Two collectors, one client
//!
//! [`PxWebCollector`] reads ONE cube — the numbers. [`PxWebTableTree`] reads the
//! CATALOGUE: the navigation tree a PxWeb host serves at the same base url, walked
//! level by level until every leaf is a table. Both go through the same
//! [`CollectorCfg::http`] core with the same 3 req/s cap, and the tree's rows are what
//! makes the cube collector configurable — before it, choosing a cube meant knowing a
//! path like `NV/NV0101/FDBR07N` by heart or by browsing SCB's website.
//!
//! # Lane EC O6: the aggregates, and the ceiling they are
//!
//! The Företagsregistret half of O6 is on PxWeb AGGREGATES — SNI × storleksklass × år,
//! never an organisationsnummer — and that is a permanent property of this API, not a
//! temporary state: the per-company register is a different, credential-gated SCB
//! service ([`crate::rest::SCB_FORETAGSREGISTRET_*`](crate::rest::SCB_FORETAGSREGISTRET_KEY_ENV)).
//! [`SCB_FDBR07N_URL`] and [`scb_foretag_sni_storleksklass_selection`] name the one
//! aggregate cube that answers "how many firms, by industry and size, by year", so the
//! ceiling is reachable in one call rather than reconstructed by each caller.
use anyhow::{Result, anyhow};
use arrow_array::RecordBatch;
use arrow_schema::{DataType, Field, Schema};
use super::arrow_util::{i64_col, str_col};
use super::arrow_util::f64_col;
use super::{
AuthKind, BatchSink, Bounds, CollectStats, Collector, CollectorCfg, CollectorMeta,
CollectorPath, Direction, GdprClass, SourceEndpoint, SourceKind, SourceTarget,
};
/// Read a dimension's index→code mapping. JSON-stat allows `category.index` as
/// either an object `{code: position}` or an array `[code, …]`.
fn codes_by_index(dim: &serde_json::Value, size: usize) -> Result<Vec<String>> {
let index = dim
.get("category")
.and_then(|c| c.get("index"))
.ok_or_else(|| anyhow!("dimension missing category.index"))?;
if let Some(arr) = index.as_array() {
return Ok(arr
.iter()
.map(|v| v.as_str().unwrap_or("").to_string())
.collect());
}
if let Some(obj) = index.as_object() {
let mut codes = vec![String::new(); size];
for (code, pos) in obj {
let p = pos
.as_u64()
.ok_or_else(|| anyhow!("category.index position not int"))?
as usize;
if p < size {
codes[p] = code.clone();
}
}
return Ok(codes);
}
Err(anyhow!("category.index is neither array nor object"))
}
/// Normalise a JSON-stat2 cube into one Arrow `RecordBatch`: a Utf8 code column
/// per dimension (named by the dimension id) + a nullable Float64 `value` column.
pub fn parse_jsonstat2(v: &serde_json::Value) -> Result<RecordBatch> {
let ids: Vec<String> = v
.get("id")
.and_then(|x| x.as_array())
.ok_or_else(|| anyhow!("JSON-stat2: missing `id`"))?
.iter()
.map(|s| s.as_str().unwrap_or("").to_string())
.collect();
let sizes: Vec<usize> = v
.get("size")
.and_then(|x| x.as_array())
.ok_or_else(|| anyhow!("JSON-stat2: missing `size`"))?
.iter()
.map(|s| s.as_u64().unwrap_or(0) as usize)
.collect();
if ids.len() != sizes.len() {
return Err(anyhow!("JSON-stat2: id/size length mismatch"));
}
let dimension = v
.get("dimension")
.ok_or_else(|| anyhow!("JSON-stat2: missing `dimension`"))?;
// index→code per dimension
let mut dim_codes: Vec<Vec<String>> = Vec::with_capacity(ids.len());
for (j, id) in ids.iter().enumerate() {
let dim = dimension
.get(id)
.ok_or_else(|| anyhow!("dimension `{id}` not described"))?;
dim_codes.push(codes_by_index(dim, sizes[j])?);
}
let total: usize = sizes.iter().product();
// values: array (dense) or object (sparse {flat_index: val}); nulls allowed.
let mut values: Vec<Option<f64>> = vec![None; total];
match v.get("value") {
Some(serde_json::Value::Array(arr)) => {
for (i, cell) in arr.iter().enumerate() {
if i < total {
values[i] = cell.as_f64();
}
}
}
Some(serde_json::Value::Object(obj)) => {
for (k, cell) in obj {
if let Ok(i) = k.parse::<usize>() {
if i < total {
values[i] = cell.as_f64();
}
}
}
}
_ => return Err(anyhow!("JSON-stat2: missing `value`")),
}
// Decode each flat cell index → per-dimension code (row-major, last varies fastest).
let mut code_cols: Vec<Vec<Option<String>>> = vec![Vec::with_capacity(total); ids.len()];
for i in 0..total {
let mut rem = i;
let mut idx = vec![0usize; ids.len()];
for j in (0..ids.len()).rev() {
let sz = sizes[j].max(1);
idx[j] = rem % sz;
rem /= sz;
}
for j in 0..ids.len() {
code_cols[j].push(Some(dim_codes[j].get(idx[j]).cloned().unwrap_or_default()));
}
}
// Assemble schema + columns.
let mut fields: Vec<Field> = ids
.iter()
.map(|id| Field::new(id, DataType::Utf8, false))
.collect();
fields.push(Field::new("value", DataType::Float64, true));
let schema = std::sync::Arc::new(Schema::new(fields));
let mut cols: Vec<arrow_array::ArrayRef> = code_cols.iter().map(|c| str_col(c)).collect();
cols.push(f64_col(&values));
Ok(RecordBatch::try_new(schema, cols)?)
}
/// **The PxWeb request document for a cell selection** — the POST body, as one pure
/// function.
///
/// It exists because the thing that was wrong here was invisible to every test in the
/// crate. `collect` used to send `{"selection": …}`, PxWeb requires `{"query": …}`, and
/// SCB answers a body keyed `selection` with **HTTP 400 Bad Request** and a bare text
/// body — no JSON, no field name. Measured 2026-09-05 against
/// `NV/NV0101/FDBR07N`: the identical selection is 400 under `selection` and 200 under
/// `query`. The only test that existed asserted [`parse_jsonstat2`] over a canned cube —
/// the SHAPER, never the REQUEST — so a POST path that had never once succeeded was
/// green.
///
/// The cfg key stays `params.selection`: that is what an operator writes and what it
/// means. This is the one place the two names are reconciled.
pub fn pxweb_request_document(selection: &serde_json::Value) -> serde_json::Value {
serde_json::json!({ "query": selection, "response": { "format": "json-stat2" } })
}
/// The PxWeb / JSON-stat2 collector. The emitted Arrow schema is cube-shaped, so
/// it is derived from the configured cube's dimension list.
#[derive(Default)]
pub struct PxWebCollector;
impl PxWebCollector {
/// Build the cube schema for a configured dimension list (`params.dims`),
/// falling back to a generic `dim0..` shape when dims aren't pre-declared.
fn cube_schema(cfg: &CollectorCfg) -> Schema {
let dims: Vec<String> = cfg
.params
.get("dims")
.and_then(|d| d.as_array())
.map(|a| {
a.iter()
.filter_map(|v| v.as_str().map(str::to_string))
.collect()
})
.unwrap_or_else(|| vec!["region".into(), "year".into(), "ContentsCode".into()]);
let mut fields: Vec<Field> = dims
.iter()
.map(|d| Field::new(d, DataType::Utf8, false))
.collect();
fields.push(Field::new("value", DataType::Float64, true));
Schema::new(fields)
}
}
#[async_trait::async_trait]
impl Collector for PxWebCollector {
fn meta(&self) -> CollectorMeta {
CollectorMeta {
name: "pxweb-jsonstat2".into(),
source_kind: SourceKind::PxWeb,
path: CollectorPath::RustNative,
direction: Direction::Source,
formats: vec!["json-stat2".into(), "json".into()],
license: "open (CC0 — SCB et al.)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None,
join_keys: vec!["kommunkod".into(), "year".into(), "elområde".into()],
description: "PxWeb 2.0 / JSON-stat2 statistical cubes → Arrow (SCB, \
Energimyndigheten, BRÅ, Folkhälsomyndigheten, …)."
.into(),
}
}
fn source_target(&self) -> SourceTarget {
SourceTarget::warehouse(
SourceEndpoint::PxWebApi,
"https://api.scb.se/OV0104/v1/doris/sv/ssd",
"pxweb_cube",
)
}
/// **One client, many agencies — so the card names the one it points AT.**
///
/// PxWeb is a product, not an authority: SCB, Energimyndigheten, BRÅ and
/// Folkhälsomyndigheten each run their own instance and each publish their own
/// terms. The declared `source_target()` endpoint is SCB's Statistical Database,
/// so SCB's published figures are the ones quoted here — and the first quirk says
/// out loud that pointing `base_url` at another agency moves the terms with it.
///
/// The rate limit is not a guess and not new knowledge: SCB publishes **30 calls
/// per 10 seconds and 150 000 data cells per call, per IP address**, which is the
/// same fact [`collect`](Collector::collect) already encodes as
/// `HttpCore::new(3)`. Fetched from SCB's own PxWebApi page on 2026-09-04.
fn human_doc(&self) -> Option<crate::human_doc::HumanDoc> {
use crate::human_doc::{Contact, HumanDoc, RateLimit};
Some(
HumanDoc::new(
"Statistics Sweden (SCB) for the declared endpoint \
(`api.scb.se/OV0104/v1/doris`) — the same client serves any PxWeb host, and \
`base_url` chooses which agency answers",
"the agency that publishes the cube: SCB owns the Statistical Database (SSD) \
tables the declared endpoint serves, and Energimyndigheten / BRÅ / \
Folkhälsomyndigheten own theirs",
"https://www.scb.se/en/services/open-data-api/",
"https://www.scb.se/en/services/open-data-api/api-for-the-statistical-database/",
)
.with_contact(Contact::Form(
"https://www.scb.se/en/About-us/contact-us/statistikservice/ask-us/".into(),
))
.with_rate_limit(RateLimit::Documented(
"SCB PxWebApi: a maximum of 30 calls per 10 seconds AND a limit of 150 000 data \
cells per call, both per IP address"
.into(),
))
.with_quirk(
"Everything below the authority line is SCB's. Point `base_url` at another \
PxWeb instance and the transport is identical but the licence, the caps and the \
support desk all change — re-read that agency's own terms before you schedule \
it.",
)
.with_quirk(
"SCB's licence is CC0, in their words: \"you may use and disseminate or provide \
such data without any requirement to state the source\". That is unusually \
permissive and does NOT carry over to other PxWeb hosts.",
)
.with_quirk(
"The 150 000-cell cap is per CALL, so a wide selection has to be split by the \
caller: `collect` pushes exactly one batch per call and does not paginate a \
too-large cube for you.",
)
.with_quirk(
"`collect` builds `HttpCore::new(3)` — 3 requests/second, deliberately under \
SCB's 30-per-10-seconds so that a burst can never trip the per-IP cap. That \
number is a floor chosen against SCB's limit, not a property of PxWeb.",
)
.with_quirk(
"`params.selection` decides the HTTP VERB: present → POST the JSON-stat2 \
selection to `base_url`; absent → a plain GET of `base_url`. A table URL that \
only answers POST will silently return the wrong document under the GET path.",
)
.with_quirk(
"Only dimension CODES are emitted — [`parse_jsonstat2`] reads `category.index` \
and never `category.label`, so a `region` column carries \"0180\", not \
\"Stockholm\". Resolve labels downstream or fetch the codelist separately.",
)
.with_quirk(
"Row count is always the PRODUCT of `size`, because a missing cell stays a null \
`value` rather than being dropped (RAGNARÖK: an empty cell is honestly null). A \
mostly-empty cube therefore costs its full dense width.",
)
.with_quirk(
"`value` arrives as either a dense array or a sparse `{flat_index: value}` \
object, and `category.index` as either an array or a `{code: position}` object. \
All four combinations decode to the same row order — but a code whose declared \
position is beyond `size` is dropped silently.",
),
)
}
/// **Every DIMENSION code column of this cube** — knowable only from the cfg.
///
/// A JSON-stat2 observation is identified by its full coordinate: one code per
/// dimension (`region`, `year`, `ContentsCode`, and whatever else the table
/// declares). The derived key was `["kommunkod", "year", "elområde"]` — the words
/// the JOIN SPINE uses, none of which is a column this collector emits.
///
/// Read from the same [`cube_schema`](Self::cube_schema) the schema is built from,
/// so a table with more dimensions keys on more of them without a second edit.
fn business_key_for(&self, cfg: &CollectorCfg) -> crate::history::BusinessKey {
let dims: Vec<String> = Self::cube_schema(cfg)
.fields()
.iter()
.map(|f| f.name().clone())
.filter(|n| n != "value")
.collect();
if dims.is_empty() {
return crate::history::BusinessKey::append_only(
"the cube declared no dimensions, so an observation has no coordinate",
);
}
let cols: Vec<&str> = dims.iter().map(String::as_str).collect();
crate::history::BusinessKey::of(
&cols,
format!(
"one observation per full cube coordinate ({}) — the dimension set is \
table-driven, so the key is read from the same cube schema",
dims.join(", ")
),
)
}
/// **A regional statistics cube as a graph** (lane EC O7).
///
/// The observation has no key of its own — its identity is the full cube
/// coordinate, which is message-driven, so it is DERIVED from every column except
/// the measure (the same rule `business_key_for` uses; one definition of "the
/// cube's key", read two ways).
fn graph_mapping(&self) -> Option<crate::graph_mapping::GraphMapping> {
use crate::graph_mapping::{DerivedId, GraphMapping, SpineLookup};
Some(
GraphMapping::new(
"scb_regions",
"SCB PxWeb — regionala tal",
"scb_cube",
"the cube's `region` column is a code whose GRANULARITY is the source \
table's, not a declared field: 4 characters is a kommun, 2 a län, `00` \
is Riket and gets no node. A code that is in neither list resolves to \
NULL and drops its edge.",
)
.derive_id(DerivedId::from_all_except("obs_id", &["value"]))
.with_spine("region", SpineLookup::ByCode)
.vertex("ScbObservation", "obs_id", &["region", "year", "value"])
.edge(
"FOR_MUNICIPALITY",
("ScbObservation", "obs_id", "obs_id"),
("Municipality", "municipality_code", "kommun_code"),
)
.edge(
"FOR_COUNTY",
("ScbObservation", "obs_id", "obs_id"),
("County", "county_code", "county_code"),
)
.edge("IN_PERIOD", ("ScbObservation", "obs_id", "obs_id"), ("Year", "year", "year")),
)
}
fn schema(&self, cfg: &CollectorCfg) -> Result<Schema> {
Ok(Self::cube_schema(cfg))
}
async fn collect(
&self,
cfg: &CollectorCfg,
_bounds: Bounds,
sink: &BatchSink,
) -> Result<CollectStats> {
let http = cfg.http(3); // SCB: ≤30 calls / 10s → be gentle
let auth = cfg.auth.resolve_async().await?;
// POST a JSON selection when `params.selection` is present, else GET the cube.
//
// **The wire key is `query`, not `selection`.** The cfg key stays
// `params.selection` — that is what an operator writes and what it MEANS — but
// PxWeb's own request document calls the array `query`, and a body keyed
// `selection` answers **HTTP 400 Bad Request** with a bare text body, not a JSON
// error naming the field. Measured against
// `api.scb.se/OV0104/v1/doris/sv/ssd/NV/NV0101/FDBR07N` on 2026-09-05: the
// identical selection is 400 under `selection` and 200 under `query`.
//
// This was wrong from the first version of this collector and nothing caught it,
// because the only test that existed asserted `parse_jsonstat2` over a canned
// cube — the SHAPER, never the REQUEST. A POST path that has never once
// succeeded is exactly the kind of green LAW 2 exists to disbelieve, and it took
// running the thing at SCB to see it.
let body = if let Some(sel) = cfg.params.get("selection") {
let req = pxweb_request_document(sel);
http.post_text(
&cfg.base_url,
req.to_string(),
"application/json",
auth.as_ref(),
)
.await?
} else {
http.get_text(&cfg.base_url, auth.as_ref()).await?
};
let json: serde_json::Value = serde_json::from_str(&body)?;
let rb = parse_jsonstat2(&json)?;
let n = rb.num_rows() as u64;
if n == 0 {
return Ok(CollectStats {
reason: Some("PxWeb cube had no cells".into()),
..Default::default()
});
}
sink.push(rb)?;
Ok(CollectStats {
rows: n,
batches: 1,
table: cfg.table.clone(),
reason: None,
})
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The CATALOGUE — a PxWeb host's table tree
// ─────────────────────────────────────────────────────────────────────────────
/// **SCB's Statistical Database (SSD) root** — the tree every other SCB path hangs off.
/// Measured live 2026-09-05: HTTP 200, **22** subject areas.
pub const SCB_SSD_ROOT: &str = "https://api.scb.se/OV0104/v1/doris/sv/ssd";
/// **The SNI × storleksklass × år company-count cube** — lane EC O6's aggregate half.
///
/// `NV/NV0101/FDBR07N`, *"Företag (FDB), antal efter näringsgren SNI 2007,
/// storleksklass, tabellinnehåll och år"*. Measured live 2026-09-05 from its own
/// metadata: **1 823** SNI 2007 codes × **9** size classes × **1** measure × **18** years
/// (2008–2025) = **295 326** cells, last updated `2025-12-02T08:00:00`.
///
/// This is the ceiling the owner named, stated as a fact rather than a limitation: it
/// answers how many firms exist by industry and by size, per year, for the whole
/// country — and it can never answer anything about a NAMED company, because no
/// organisationsnummer exists anywhere in it.
pub const SCB_FDBR07N_URL: &str = "https://api.scb.se/OV0104/v1/doris/sv/ssd/NV/NV0101/FDBR07N";
/// **SCB's published per-call cell cap, and what breaching it actually looks like.**
///
/// SCB documents 150 000 data cells per call. [`SCB_FDBR07N_URL`] holds 295 326, so
/// `filter: all` on every dimension is over the cap — and the answer is **HTTP 403 with
/// a 1 233-byte IIS HTML error page**, not a JSON error and not a 400 (measured
/// 2026-09-05). A collector that assumed a JSON body would report a parse failure and an
/// operator would go looking for a malformed cube.
pub const SCB_CELL_CAP: usize = 150_000;
/// **The lane's own selection over [`SCB_FDBR07N_URL`]** — the 22 top-level SNI
/// divisions (`A-U` plus the 21 lettered sections) × all 9 size classes × all 18 years.
///
/// **3 564 cells**, measured live 2026-09-05: HTTP 200, 18 085 bytes, 3 564 values, zero
/// nulls. One writer for the selection, because the reason it is 22 SNI codes and not
/// 1 823 is [`SCB_CELL_CAP`], and a caller who re-derived it would re-derive the 403 too.
///
/// The `ContentsCode` is pinned to `000002YB` (*Antal*, the count of firms) rather than
/// left as `all`, because the cube declares exactly one measure today and an `all` that
/// silently widened when SCB adds a second is how a cell budget stops being a budget.
pub fn scb_foretag_sni_storleksklass_selection() -> serde_json::Value {
serde_json::json!([
{"code": "SNI2007", "selection": {"filter": "item", "values": [
"A-U", "A", "B", "C", "D", "E", "F", "G", "H", "I", "J",
"K", "L", "M", "N", "O", "P", "Q", "R", "S", "T", "U"
]}},
{"code": "Storleksklass", "selection": {"filter": "all", "values": ["*"]}},
{"code": "ContentsCode", "selection": {"filter": "item", "values": ["000002YB"]}},
{"code": "Tid", "selection": {"filter": "all", "values": ["*"]}}
])
}
/// The dimension names [`scb_foretag_sni_storleksklass_selection`] produces, in the
/// order JSON-stat2 returns them — the `params.dims` a [`PxWebCollector`] cfg needs so
/// its schema and its `business_key_for` describe the cube it will actually fetch.
pub const SCB_FDBR07N_DIMS: &[&str] = &["SNI2007", "Storleksklass", "ContentsCode", "Tid"];
/// **A ready `CollectorCfg` for the SNI × storleksklass × år cube** — url, selection and
/// dims in one place.
///
/// The dims matter as much as the selection. [`PxWebCollector::business_key_for`] reads
/// `params.dims` to decide what identifies an observation, and a cfg that supplied the
/// selection but not the dims would key the cube on the fallback
/// `region · year · ContentsCode` — three columns this cube does not have, which is
/// exactly the class of defect lane EC O3 found 29 of.
pub fn scb_foretag_sni_storleksklass_cfg() -> CollectorCfg {
CollectorCfg::new(SCB_FDBR07N_URL, "scb_foretag_sni_storleksklass").with_params(
serde_json::json!({
"selection": scb_foretag_sni_storleksklass_selection(),
"dims": SCB_FDBR07N_DIMS,
}),
)
}
/// One node of a PxWeb navigation listing, flattened.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TreeNode {
/// The path BELOW the walk's root, `/`-joined (`NV/NV0101/FDBR07N`). Unique within
/// one host, and the business key.
pub path: String,
/// The containing folder's path — empty at the root level.
pub parent_path: String,
/// The node's own id (`FDBR07N`).
pub id: String,
/// `l` for a folder (another listing lives below it), `t` for a table.
pub node_type: String,
/// The human title PxWeb serves.
pub text: String,
/// The table's own last-updated stamp. **Folders do not have one** — PxWeb serves
/// `updated` on `t` nodes only — so this is NULL for every `l` row, which is a fact
/// about the source and not a gap in the walk.
pub updated: Option<String>,
/// How deep below the root, 1-based.
pub depth: i64,
}
/// **Flatten one PxWeb listing into nodes** — the offline contract for the tree walk.
///
/// A listing is a bare array of `{id, type, text, updated?}`. Anything that is not an
/// object carrying an `id` is SKIPPED rather than landed as a row of nulls: PxWeb hosts
/// differ, and an element with no id is not a node anything could navigate to.
pub fn parse_listing(
v: &serde_json::Value,
parent_path: &str,
depth: i64,
) -> Result<Vec<TreeNode>> {
let arr = v
.as_array()
.ok_or_else(|| anyhow!("PxWeb listing is not an array"))?;
let mut out = Vec::with_capacity(arr.len());
for n in arr {
let Some(id) = n.get("id").and_then(|x| x.as_str()) else {
continue;
};
let node_type = n
.get("type")
.and_then(|x| x.as_str())
.unwrap_or_default()
.to_string();
let path = if parent_path.is_empty() {
id.to_string()
} else {
format!("{parent_path}/{id}")
};
out.push(TreeNode {
path,
parent_path: parent_path.to_string(),
id: id.to_string(),
node_type,
text: n
.get("text")
.and_then(|x| x.as_str())
.unwrap_or_default()
.to_string(),
updated: n
.get("updated")
.and_then(|x| x.as_str())
.map(str::to_string),
depth,
});
}
Ok(out)
}
/// **The PxWeb TABLE TREE collector** — one row per node of a host's navigation tree.
///
/// The catalogue half of the PxWeb client (lane EC O6). [`PxWebCollector`] reads one
/// cube and has to be TOLD which; this reads what there is to choose from, which is a
/// thing korp could not answer at all before it.
///
/// **It is a WALK, not a fetch.** A PxWeb listing names its children and nothing more,
/// so reaching the tables means one request per folder. SCB's SSD is three levels —
/// subject area → statistics product → table — and `params.max_depth` bounds it; a host
/// with a deeper tree lands what fits and SAYS the walk was cut short.
#[derive(Default)]
pub struct PxWebTableTree;
/// The default depth [`PxWebTableTree`] walks. SCB's SSD is subject area → product →
/// table, and one more level costs nothing on a host that has no fourth: the walk only
/// descends into nodes that actually declared themselves folders.
pub const PXWEB_TREE_DEFAULT_DEPTH: i64 = 4;
impl PxWebTableTree {
/// The emitted schema — FIXED, because a navigation node has the same shape on
/// every PxWeb host. (Unlike a cube, whose columns are the table's own dimensions
/// and are therefore knowable only from the cfg.)
fn tree_schema() -> Schema {
Schema::new(vec![
Field::new("path", DataType::Utf8, false),
Field::new("parent_path", DataType::Utf8, true),
Field::new("id", DataType::Utf8, false),
Field::new("node_type", DataType::Utf8, true),
Field::new("text", DataType::Utf8, true),
Field::new("updated", DataType::Utf8, true),
Field::new("depth", DataType::Int64, true),
])
}
/// Nodes → one Arrow batch. Separate from the walk so the SHAPING is assertable
/// without a network.
pub fn batch(nodes: &[TreeNode]) -> Result<RecordBatch> {
let schema = std::sync::Arc::new(Self::tree_schema());
let cols: Vec<arrow_array::ArrayRef> = vec![
str_col(&nodes.iter().map(|n| Some(n.path.clone())).collect::<Vec<_>>()),
// A ROOT node has no parent, and that is NULL — never `""`. The graph
// mapping's CONTAINED_IN edge is dropped on a NULL endpoint, which is how
// "this subject area is contained in nothing" stays true by construction.
str_col(
&nodes
.iter()
.map(|n| (!n.parent_path.is_empty()).then(|| n.parent_path.clone()))
.collect::<Vec<_>>(),
),
str_col(&nodes.iter().map(|n| Some(n.id.clone())).collect::<Vec<_>>()),
str_col(
&nodes
.iter()
.map(|n| Some(n.node_type.clone()))
.collect::<Vec<_>>(),
),
str_col(&nodes.iter().map(|n| Some(n.text.clone())).collect::<Vec<_>>()),
str_col(&nodes.iter().map(|n| n.updated.clone()).collect::<Vec<_>>()),
i64_col(&nodes.iter().map(|n| Some(n.depth)).collect::<Vec<_>>()),
];
Ok(RecordBatch::try_new(schema, cols)?)
}
}
#[async_trait::async_trait]
impl Collector for PxWebTableTree {
fn meta(&self) -> CollectorMeta {
CollectorMeta {
name: "pxweb-tables".into(),
source_kind: SourceKind::PxWeb,
path: CollectorPath::RustNative,
direction: Direction::Source,
formats: vec!["json".into()],
license: "open (CC0 — SCB et al.; the catalogue carries its host's own terms)"
.into(),
gdpr: GdprClass::Open,
auth: AuthKind::None,
// The path IS the join key: it is what a `pxweb-jsonstat2` cfg's `base_url`
// ends with, so a catalogue row and the cube fetched from it join on a
// column both sides really carry.
join_keys: vec!["path".into()],
description: "PxWeb navigation TREE → one row per subject area, statistics \
product and table (SCB SSD: 22 subject areas at the root, \
measured 2026-09-05). The catalogue that makes \
`pxweb-jsonstat2` configurable."
.into(),
}
}
fn source_target(&self) -> SourceTarget {
SourceTarget::warehouse(SourceEndpoint::PxWebApi, SCB_SSD_ROOT, "pxweb_tables")
}
/// **One row per node, keyed on its PATH.**
///
/// Not `id`: a PxWeb id is unique only among its siblings, so `NV0101` identifies a
/// node only once you know it sits under `NV`. The path is also exactly what a
/// `pxweb-jsonstat2` cfg's `base_url` ends with, which makes it the column a landed
/// cube joins back on — one string doing both jobs.
fn business_key(&self) -> crate::history::BusinessKey {
crate::history::BusinessKey::of(
&["path"],
"one row per navigation node, keyed on its full path below the walk root — \
a PxWeb `id` is unique only among its siblings, and the path is also what a \
`pxweb-jsonstat2` cfg's `base_url` ends with",
)
}
/// **Every read re-walks the whole tree** — 24 h, and the interval IS the cost.
///
/// There is nothing else honest to declare. A PxWeb listing serves no validator korp
/// could send back, and even if the ROOT answered `304` that would say nothing about
/// the 22 listings below it: the tree's state is the union of every node, and the
/// only way to learn it is to ask every folder. So this is
/// [`AlwaysFull`](crate::freshness::Freshness::AlwaysFull), which puts it on the
/// 24-hour default rather than the 5-minute one — the right cadence for a catalogue
/// that changes when an agency publishes a new statistical product.
fn freshness(&self) -> crate::freshness::Freshness {
crate::freshness::Freshness::AlwaysFull {
why: "a PxWeb listing serves no validator, and a 304 on the root would say \
nothing about the folders below it — the tree's state is the union of \
every node, so every run re-walks all of them (one request per folder)"
.into(),
}
}
/// **A catalogue as a graph** (lane EC O7): the containment the tree IS.
///
/// One label, one self-edge. A folder and a table are the same KIND of thing here —
/// a node with a path and a title — and `node_type` is the property that tells them
/// apart, so splitting them into two labels would put in the schema a distinction
/// the source keeps in a field.
fn graph_mapping(&self) -> Option<crate::graph_mapping::GraphMapping> {
use crate::graph_mapping::GraphMapping;
Some(
GraphMapping::new(
"pxweb_tables",
"PxWeb — tabellträdet",
"pxweb_tables",
"a root-level node has no parent, so its `parent_path` is NULL and the \
sink drops its CONTAINED_IN edge — the tree lands as 22 disconnected \
components at SCB, one per subject area, which is what the source \
actually is. The path is also unique only WITHIN one PxWeb host: two \
instances walked into the same table would MERGE nodes that share a \
path and mean different things.",
)
.vertex(
"PxWebNode",
"path",
&["id", "node_type", "text", "updated", "depth"],
)
.edge(
"CONTAINED_IN",
("PxWebNode", "path", "path"),
("PxWebNode", "parent_path", "path"),
),
)
}
fn human_doc(&self) -> Option<crate::human_doc::HumanDoc> {
use crate::human_doc::{Contact, HumanDoc, RateLimit};
Some(
HumanDoc::new(
"Statistics Sweden (SCB) for the declared root \
(`api.scb.se/OV0104/v1/doris/sv/ssd`) — the same walk works against any \
PxWeb host, and `base_url` chooses which agency answers",
"the agency that publishes the catalogue: SCB owns the SSD tree the \
declared root serves, and Energimyndigheten / BRÅ / \
Folkhälsomyndigheten own theirs",
"https://www.scb.se/en/services/open-data-api/",
"https://www.scb.se/en/services/open-data-api/api-for-the-statistical-database/",
)
.with_contact(Contact::Form(
"https://www.scb.se/en/About-us/contact-us/statistikservice/ask-us/".into(),
))
.with_rate_limit(RateLimit::Documented(
"SCB PxWebApi: a maximum of 30 calls per 10 seconds per IP address. This \
collector is a WALK — one request per folder — so it is the collector \
that call budget actually binds"
.into(),
))
.with_quirk(
"MEASURED 2026-09-05: the SSD root serves 22 subject areas, and the tree \
below them is three levels (area → statistics product → table). A full \
walk is ~1 + 22 + one request per statistics product, at 3 req/s",
)
.with_quirk(
"`updated` is served on TABLE (`type: t`) nodes only — a folder row's \
`updated` is NULL because PxWeb has none for it, not because the walk \
missed it",
)
.with_quirk(
"a node `id` is unique only among its SIBLINGS. The row's identity is \
`path`, and `path` is host-relative: the same path on two PxWeb \
instances is two different tables",
)
.with_quirk(
"the walk descends into `type: l` nodes ONLY. A table's url does not \
serve a listing — it serves the cube's METADATA — so a walk that \
followed every node would abort on the first table it reached",
)
.with_quirk(
"the tree is the catalogue, NOT the data. A `t` row's `path` is exactly \
what a `pxweb-jsonstat2` cfg's `base_url` ends with, which is the whole \
point — but fetching that cube is a second call, with its own 150 000 \
cell cap (see `SCB_CELL_CAP`: breaching it answers HTTP 403 with an \
HTML page, not a JSON error)",
),
)
}
fn schema(&self, _cfg: &CollectorCfg) -> Result<Schema> {
Ok(Self::tree_schema())
}
async fn collect(
&self,
cfg: &CollectorCfg,
_bounds: Bounds,
sink: &BatchSink,
) -> Result<CollectStats> {
let http = cfg.http(3); // SCB: ≤30 calls / 10s — and this one is a WALK
let auth = cfg.auth.resolve_async().await?;
let max_depth = cfg
.params
.get("max_depth")
.and_then(serde_json::Value::as_i64)
.unwrap_or(PXWEB_TREE_DEFAULT_DEPTH)
.max(1);
let root = cfg.base_url.trim_end_matches('/').to_string();
let mut nodes: Vec<TreeNode> = Vec::new();
// (url, path-below-root, depth).
let mut frontier: Vec<(String, String, i64)> = vec![(root.clone(), String::new(), 1)];
let mut requests = 0usize;
let mut cut_short = false;
// Folders that declared themselves folders and then refused to open. See below.
let mut refused: Vec<String> = Vec::new();
while let Some((url, parent, depth)) = frontier.pop() {
// **A folder that refuses must not cost the other 1 500 tables.**
//
// A PxWeb node's `type` is the host's own claim about itself, and the host
// can be wrong: `NV/NV0802/NV0802Z` ("Äldre tabeller som inte uppdateras")
// is typed `l` and answers **HTTP 400 Bad Request** when asked for its
// listing, while its siblings `NV0802A`…`NV0802D` answer 200 (measured
// 2026-09-05). The `?` that used to be here aborted the entire walk on that
// one node — the whole SCB catalogue lost to one stale folder, with the
// error naming a url and nothing explaining why the catalogue was empty.
//
// So a failed branch is RECORDED and skipped. The ROOT is the exception: if
// the walk's own starting url does not answer there is no catalogue at all,
// and reporting a cheerful zero would be the silent green this crate's
// RAGNARÖK law forbids.
let fetched = http.get_text(&url, auth.as_ref()).await;
requests += 1;
let level = match fetched.and_then(|body| {
let v: serde_json::Value = serde_json::from_str(&body)?;
parse_listing(&v, &parent, depth)
}) {
Ok(level) => level,
Err(e) if !nodes.is_empty() => {
refused.push(format!("{parent} ({e})"));
continue;
}
Err(e) => return Err(e.context(format!("the walk root {url} did not answer"))),
};
for n in &level {
// Only a FOLDER has a listing below it. Asking a table's url for one
// returns the cube's METADATA — an object, not an array — which would
// abort the whole walk, so the recursion follows `type == "l"` and
// nothing else.
if n.node_type == "l" {
if depth < max_depth {
frontier.push((format!("{root}/{}", n.path), n.path.clone(), depth + 1));
} else {
cut_short = true;
}
}
}
nodes.extend(level);
}
if nodes.is_empty() {
return Ok(CollectStats {
table: cfg.table.clone(),
reason: Some(format!(
"the PxWeb listing at {root} named no nodes — this url has to be a \
PxWeb host root or a folder, never a table"
)),
..Default::default()
});
}
let tables = nodes.iter().filter(|n| n.node_type == "t").count();
let rb = Self::batch(&nodes)?;
let n = rb.num_rows() as u64;
sink.push(rb)?;
Ok(CollectStats {
rows: n,
batches: 1,
table: cfg.table.clone(),
// A walk that stopped at its depth bound, or skipped a folder that refused,
// landed rows AND is incomplete. Both are reported even though the run
// succeeded, for the same reason a truncated page crawl is: a partial
// catalogue that claims to be whole is worse than a failure, because
// nothing downstream can tell.
reason: crate::rest::join_reasons(&[
cut_short.then(|| {
format!(
"the walk stopped at max_depth={max_depth} with folders still \
unopened — this catalogue is INCOMPLETE ({tables} tables from \
{requests} requests). Raise `params.max_depth`."
)
}),
(!refused.is_empty()).then(|| {
format!(
"{} folder(s) declared themselves folders and then REFUSED to \
open, so their tables are missing from this catalogue: {}",
refused.len(),
refused.join("; ")
)
}),
]),
})
}
}