Exporting and Restoring a GeoServer Catalog With Python

Walk /rest/workspaces, /rest/styles and the store and feature type endpoints into one JSON document, strip every secret, and restore in dependency order: workspaces, styles, stores, feature types, layer groups. The export is a description of intent that belongs in version control; the credentials are injected at restore time from the environment.

The Core Challenge: An Export Is Not a Backup

GeoServer’s own data directory backup is a filesystem copy that captures everything including secrets and machine-specific paths, and it restores only onto a compatible version. That makes it a disaster-recovery tool and a poor fit for the thing teams actually want: a reviewable description of what is published, applied to whatever environment needs it.

A REST export is that description. It is text, so it diffs; it is structured, so it can be templated per environment; and it deliberately omits store passwords, which is what makes it safe to commit. What it also omits is the tile cache — a restored service is cold and re-seeds on demand — and that is a property to plan around rather than a defect.

Production-Ready Code

from __future__ import annotations

import json
import os
from dataclasses import dataclass, field
from typing import Any

import requests
from requests.auth import HTTPBasicAuth

SECRET_KEYS = {"passwd", "password", "PASSWORD"}

@dataclass
class Catalog:
    base: str
    user: str
    password: str
    timeout: int = 60

    @property
    def auth(self) -> HTTPBasicAuth:
        return HTTPBasicAuth(self.user, self.password)

    def get(self, path: str) -> dict[str, Any]:
        resp = requests.get(f"{self.base.rstrip('/')}{path}",
                            auth=self.auth, timeout=self.timeout,
                            headers={"Accept": "application/json"})
        if resp.status_code == 404:
            return {}
        resp.raise_for_status()
        return resp.json()

    def send(self, method: str, path: str, payload: Any,
             content_type: str = "application/json") -> int:
        kwargs: dict[str, Any] = {"headers": {"Content-Type": content_type}}
        kwargs["data"] = (json.dumps(payload) if content_type.endswith("json")
                          else payload)
        resp = requests.request(method, f"{self.base.rstrip('/')}{path}",
                                auth=self.auth, timeout=self.timeout, **kwargs)
        # 409 means it already exists — the desired state is reached either way.
        if resp.status_code not in (200, 201, 409):
            resp.raise_for_status()
        return resp.status_code

def _strip_secrets(node: Any) -> Any:
    """Replace every credential with a placeholder naming its env variable."""
    if isinstance(node, dict):
        out = {}
        for key, value in node.items():
            if key in SECRET_KEYS:
                out[key] = "${GEOSERVER_STORE_PASSWORD}"
            else:
                out[key] = _strip_secrets(value)
        return out
    if isinstance(node, list):
        return [_strip_secrets(v) for v in node]
    return node

def _as_list(payload: Any, outer: str, inner: str) -> list[dict[str, Any]]:
    """GeoServer returns '' for an empty collection and a dict otherwise."""
    block = (payload or {}).get(outer)
    if not block:
        return []
    items = block.get(inner, [])
    return items if isinstance(items, list) else [items]

def export_catalog(cat: Catalog) -> dict[str, Any]:
    export: dict[str, Any] = {"workspaces": [], "styles": [], "layerGroups": []}

    for style in _as_list(cat.get("/rest/styles.json"), "styles", "style"):
        body = requests.get(f"{cat.base}/rest/styles/{style['name']}.sld",
                            auth=cat.auth, timeout=cat.timeout)
        export["styles"].append({"name": style["name"],
                                 "sld": body.text if body.ok else None})

    for ws in _as_list(cat.get("/rest/workspaces.json"), "workspaces", "workspace"):
        name = ws["name"]
        entry: dict[str, Any] = {"name": name, "dataStores": []}
        stores = _as_list(cat.get(f"/rest/workspaces/{name}/datastores.json"),
                          "dataStores", "dataStore")
        for store in stores:
            detail = cat.get(f"/rest/workspaces/{name}/datastores/{store['name']}.json")
            feature_types = _as_list(
                cat.get(f"/rest/workspaces/{name}/datastores/{store['name']}"
                        f"/featuretypes.json"), "featureTypes", "featureType")
            entry["dataStores"].append({
                "definition": _strip_secrets(detail),
                "featureTypes": [
                    cat.get(f"/rest/workspaces/{name}/datastores/{store['name']}"
                            f"/featuretypes/{ft['name']}.json")
                    for ft in feature_types],
            })
        export["workspaces"].append(entry)

    for group in _as_list(cat.get("/rest/layergroups.json"), "layerGroups", "layerGroup"):
        export["layerGroups"].append(
            cat.get(f"/rest/layergroups/{group['name']}.json"))

    return export

def _resolve(node: Any, secrets: dict[str, str]) -> Any:
    """Substitute ${VAR} placeholders from the environment at restore time."""
    if isinstance(node, dict):
        return {k: _resolve(v, secrets) for k, v in node.items()}
    if isinstance(node, list):
        return [_resolve(v, secrets) for v in node]
    if isinstance(node, str) and node.startswith("${") and node.endswith("}"):
        key = node[2:-1]
        if key not in secrets:
            raise KeyError(f"{key} is required by the export but not set")
        return secrets[key]
    return node

def restore_catalog(cat: Catalog, export: dict[str, Any],
                    secrets: dict[str, str] | None = None) -> list[str]:
    """Apply an export in dependency order. Idempotent: re-running is safe."""
    secrets = secrets or dict(os.environ)
    log: list[str] = []

    for style in export.get("styles", []):
        if not style.get("sld"):
            continue
        cat.send("POST", f"/rest/styles?name={style['name']}",
                 style["sld"], content_type="application/vnd.ogc.sld+xml")
        log.append(f"style {style['name']}")

    for ws in export.get("workspaces", []):
        cat.send("POST", "/rest/workspaces", {"workspace": {"name": ws["name"]}})
        log.append(f"workspace {ws['name']}")

        for store in ws.get("dataStores", []):
            definition = _resolve(store["definition"], secrets)
            store_name = definition["dataStore"]["name"]
            cat.send("POST", f"/rest/workspaces/{ws['name']}/datastores", definition)
            log.append(f"store {ws['name']}:{store_name}")

            for feature_type in store.get("featureTypes", []):
                cat.send("POST",
                         f"/rest/workspaces/{ws['name']}/datastores/{store_name}"
                         f"/featuretypes", feature_type)
                log.append(f"layer {ws['name']}:{feature_type['featureType']['name']}")

    for group in export.get("layerGroups", []):
        cat.send("POST", "/rest/layergroups", group)
        log.append(f"group {group['layerGroup']['name']}")

    return log

Step-by-Step Walkthrough

Styles come before layers. A feature type whose default style does not exist is rejected, so the restore order is not cosmetic. Workspaces, then styles, then stores, then feature types, then layer groups — each stage depends on everything before it and on nothing after.

Secrets become named placeholders, not blanks. Replacing a password with ${GEOSERVER_STORE_PASSWORD} keeps the export self-describing: the restore fails with a message naming the missing variable rather than creating a store that cannot connect. Blanking the field instead produces a store that is created successfully and is broken, which is the failure mode the workspace and datastore guide describes.

409 is success. Every create in the restore path treats a conflict as the desired state already existing, which is what makes the whole operation re-runnable. A restore that fails half way can simply be run again once the cause is fixed, rather than needing the target cleaned first.

GeoServer returns an empty string for empty collections. /rest/workspaces.json on a fresh instance returns {"workspaces": ""} rather than an empty list, and a single item is returned as an object rather than a one-element array. _as_list normalises both, which removes a whole family of TypeErrors from the walk.

Verification

Export, restore to a disposable instance, and diff the two catalogs:

source = Catalog("https://prod.example.org/geoserver", "admin", os.environ["PROD_PW"])
target = Catalog("http://localhost:8080/geoserver", "admin", "geoserver")

export = export_catalog(source)
open("catalog.json", "w").write(json.dumps(export, indent=2, sort_keys=True))

restore_catalog(target, export, {"GEOSERVER_STORE_PASSWORD": os.environ["LOCAL_PW"]})

def names(cat):
    return sorted(l["name"] for l in _as_list(cat.get("/rest/layers.json"),
                                              "layers", "layer"))
assert names(source) == names(target), "layer sets differ after restore"

restored: 3 styles, 2 workspaces, 2 stores, 41 layers, 2 groups
layer sets match

Committing catalog.json makes every subsequent publish a reviewable diff, which is the property that makes this worth doing at all.

Gotchas & Edge Cases

The tile cache does not travel. A restored service is cold, and for a heavily seeded layer that means the first requests are slow until it warms. Plan a re-seed for the low zoom levels, as the seeding guide describes, rather than discovering it under load.

Security configuration is a separate subsystem. Users, roles and layer access rules live outside the catalog REST endpoints used here, and they are the part most likely to differ between environments anyway. Exporting them alongside is possible but rarely what you want — see Securing GeoServer REST Credentials in Python Automation.

Version differences change the payload shape. A catalog exported from one GeoServer major version may contain fields a different version rejects. Restoring across versions is worth testing deliberately rather than assuming, and pinning the version in the container image removes the question.

An export is a point in time. If the catalog changes while the walk is running, the export is internally inconsistent — a layer referencing a store that was deleted mid-walk, for instance. Exporting from a quiescent instance, or accepting the small risk, is a choice worth making consciously.

Frequently Asked Questions

---

Back to Automating GeoServer With the Python REST API

Related

• Automating GeoServer With the Python REST API — the catalog endpoints this walks
• Environment Parity for Spatial Servers — declaring the catalog rather than exporting it
• Reloading the GeoServer Catalog and Clearing Caches From Python — the reload a restored data directory needs