Resource Data Sources

Resource data sources load tabular data from CKAN resources. They are used automatically by the Resource View feature but can also be instantiated manually when you need to build a table backed by a resource file or the Datastore.

All file-based sources extend BaseResourceDataSource, which handles source path resolution (local upload vs. remote URL) and pluggable caching. The CKAN Datastore source (DataStoreDataSource) is separate and queries the Datastore API directly without any caching.

---

File-based sources

CsvUrlDataSource

Reads a CSV file from a local path or remote URL. The delimiter is detected automatically.

from ckanext.tables.shared import CsvUrlDataSource

# From a direct URL
source = CsvUrlDataSource(url="https://example.com/data.csv")

# From a CKAN resource dict (resolves upload path automatically)
source = CsvUrlDataSource(resource=resource_dict)

Source code in ckanext/tables/data_sources.py

class CsvUrlDataSource(BaseResourceDataSource):
    @staticmethod
    def _read_csv(path: str, **kwargs: Any) -> pd.DataFrame:
        """Read a local CSV file, using the fast C engine when possible.

        Sniffs the delimiter from the first few lines so the C engine can be
        used for the actual read; only falls back to pandas' own slower
        ``engine="python"`` auto-detection when sniffing fails.
        """
        delimiter = _sniff_csv_delimiter(path)
        if delimiter is not None:
            return pd.read_csv(path, sep=delimiter, **kwargs)

        return pd.read_csv(path, sep=None, engine="python", **kwargs)

    def fetch_dataframe(self) -> pd.DataFrame:
        try:
            with self._open_source() as path:
                return self._read_csv(path)
        except Exception:
            log.exception("Error fetching CSV from %s", self.get_source_path())
            return pd.DataFrame()

    def get_columns(self) -> list[str]:
        if self._load_from_cache():
            return list(self._df.columns) if self._df is not None else []

        try:
            with self._open_source() as path:
                df_preview = self._read_csv(path, nrows=0)
        except (OSError, ValueError, pd.errors.ParserError):
            log.exception("Failed fast CSV schema read, falling back to full load")
            self._ensure_loaded()
            return list(self._df.columns) if self._df is not None else []

        return list(df_preview.columns)

---

XlsxUrlDataSource

Reads the first sheet of an Excel workbook (.xlsx).

from ckanext.tables.shared import XlsxUrlDataSource

source = XlsxUrlDataSource(url="https://example.com/report.xlsx")

Source code in ckanext/tables/data_sources.py

class XlsxUrlDataSource(BaseResourceDataSource):
    def fetch_dataframe(self) -> pd.DataFrame:
        try:
            with self._open_source() as path:
                return pd.read_excel(path)
        except Exception:
            log.exception("Error fetching XLSX from %s", self.get_source_path())
            return pd.DataFrame()

    def get_columns(self) -> list[str]:
        if self._load_from_cache():
            return list(self._df.columns) if self._df is not None else []

        try:
            with self._open_source() as path:
                df_preview = pd.read_excel(path, nrows=0)
        except (OSError, ValueError):
            log.exception("Failed fast XLSX schema read, falling back to full load")
            self._ensure_loaded()
            return list(self._df.columns) if self._df is not None else []

        return list(df_preview.columns)

---

OrcUrlDataSource

Reads an Apache ORC columnar file.

from ckanext.tables.shared import OrcUrlDataSource

source = OrcUrlDataSource(url="https://example.com/data.orc")

Source code in ckanext/tables/data_sources.py

class OrcUrlDataSource(BaseResourceDataSource):
    def fetch_dataframe(self) -> pd.DataFrame:
        try:
            with self._open_source() as path:
                return pd.read_orc(path)
        except Exception:
            log.exception("Error fetching ORC from %s", self.get_source_path())
            return pd.DataFrame()

    def get_columns(self) -> list[str]:
        if self._load_from_cache():
            return list(self._df.columns) if self._df is not None else []

        try:
            with self._open_source() as path:
                schema_names = orc.ORCFile(path).schema.names
        except (OSError, ValueError, pa.ArrowInvalid):
            log.exception("Failed fast ORC schema read, falling back to full load")
            self._ensure_loaded()
            return list(self._df.columns) if self._df is not None else []

        return schema_names

---

ParquetUrlDataSource

Reads an Apache Parquet columnar file.

from ckanext.tables.shared import ParquetUrlDataSource

source = ParquetUrlDataSource(url="https://example.com/data.parquet")

Source code in ckanext/tables/data_sources.py

class ParquetUrlDataSource(BaseResourceDataSource):
    def fetch_dataframe(self) -> pd.DataFrame:
        try:
            with self._open_source() as path:
                return pd.read_parquet(path)
        except Exception:
            log.exception("Error fetching Parquet from %s", self.get_source_path())
            return pd.DataFrame()

    def get_columns(self) -> list[str]:
        if self._load_from_cache():
            return list(self._df.columns) if self._df is not None else []

        try:
            with self._open_source() as path:
                schema_names = pq.read_schema(path).names
        except (OSError, ValueError, pa.ArrowInvalid):
            log.exception("Failed fast Parquet schema read, falling back to full load")
            self._ensure_loaded()
            return list(self._df.columns) if self._df is not None else []

        return schema_names

---

FeatherUrlDataSource

Reads an Apache Arrow Feather file.

from ckanext.tables.shared import FeatherUrlDataSource

source = FeatherUrlDataSource(url="https://example.com/data.feather")

Source code in ckanext/tables/data_sources.py

class FeatherUrlDataSource(BaseResourceDataSource):
    def fetch_dataframe(self) -> pd.DataFrame:
        try:
            with self._open_source() as path:
                return pd.read_feather(path)
        except Exception:
            log.exception("Error fetching Feather from %s", self.get_source_path())
            return pd.DataFrame()

    def get_columns(self) -> list[str]:
        if self._load_from_cache():
            return list(self._df.columns) if self._df is not None else []

        try:
            with self._open_source() as path:
                schema_names = feather.read_table(path, columns=[]).schema.names
        except (OSError, ValueError, pa.ArrowInvalid):
            log.exception("Failed fast Feather schema read, falling back to full load")
            self._ensure_loaded()
            return list(self._df.columns) if self._df is not None else []

        return schema_names

---

Caching

All file-based sources inherit from BaseResourceDataSource, which caches the fetched DataFrame to avoid re-downloading on every request. The cache backend and TTL are controlled globally via configuration (see Configuration):

ckanext.tables.cache.backend = feather  # or "parquet", "pickle", "redis"
ckanext.tables.cache.cache_dir = /var/cache/ckanext-tables

You can override the backend or TTL per instance:

from ckanext.tables.shared import CsvUrlDataSource, RedisCacheBackend

source = CsvUrlDataSource(
    url="https://example.com/data.csv",
    cache_backend=RedisCacheBackend(),
    cache_ttl=120,  # seconds
)

A data source that loads resource data from a file or URL.

The cache backend defaults to the value of ckanext.tables.cache.backend ("feather" by default). Pass an explicit cache_backend to override for a specific instance.

Override cache_ttl on a subclass or pass it to the constructor to change the expiry.

PARAMETER	DESCRIPTION
url	Direct URL to fetch data from. TYPE: str | None DEFAULT: None
resource	The CKAN resource dictionary. TYPE: dict[str, Any] | None DEFAULT: None
cache_backend	Override the configured cache backend for this instance. TYPE: CacheBackend | None DEFAULT: None
cache_ttl	Override the default TTL (seconds) for this instance. TYPE: int | None DEFAULT: None

Source code in ckanext/tables/data_sources.py

class BaseResourceDataSource(CachedDataSourceMixin, PandasDataSource):
    """A data source that loads resource data from a file or URL.

    The cache backend defaults to the value of ``ckanext.tables.cache.backend``
    (``"feather"`` by default). Pass an explicit *cache_backend* to
    override for a specific instance.

    Override ``cache_ttl`` on a subclass or pass it to the constructor to
    change the expiry.

    Args:
        url: Direct URL to fetch data from.
        resource: The CKAN resource dictionary.
        cache_backend: Override the configured cache backend for this instance.
        cache_ttl: Override the default TTL (seconds) for this instance.
    """

    def __init__(
        self,
        url: str | None = None,
        resource: dict[str, Any] | None = None,
        cache_backend: CacheBackend | None = None,
        cache_ttl: int | None = None,
    ):
        super().__init__()

        if not url and not resource:
            raise ValueError("Either url or resource_id must be provided")

        self.url = url
        self.resource = resource
        self._source_path: str = ""
        self._upload_storage: Any | None = None
        self.cache_backend = cache_backend if cache_backend is not None else get_cache_backend()
        self.cache_ttl = cache_ttl if cache_ttl is not None else get_cache_ttl()

    def get_cache_key(self) -> str:
        return f"resource-{self.resource['id']}" if self.resource else f"url-{self.url}"

    def get_source_path(self) -> str:
        if self._source_path:
            return self._source_path

        if self.resource:
            try:
                if self.resource.get("url_type") == "upload":
                    upload = uploader.get_resource_uploader(self.resource)
                    self._source_path = upload.get_path(self.resource["id"])

                    # On CKAN 2.12+ with file-keeper storage, get_path() returns a
                    # storage-relative Location, not a filesystem path — stash the
                    # storage so _open_source() reads through its API instead of
                    # handing that relative string straight to pandas.
                    fk_upload_cls = getattr(uploader, "FKResourceUpload", None)
                    if fk_upload_cls is not None and isinstance(upload, fk_upload_cls):
                        self._upload_storage = getattr(upload, "storage", None)

                    return self._source_path

                if self.resource.get("url"):
                    self._source_path = self._ensure_remote_url(self.resource["url"])
                    return self._source_path

            except (OSError, TypeError, tk.ValidationError, tk.ObjectNotFound):
                log.warning(
                    "Failed to resolve path for resource %s, falling back to provided url",
                    self.resource.get("id"),
                    exc_info=True,
                )

        if self.url:
            self._source_path = self._ensure_remote_url(self.url)
            return self._source_path

        raise ValueError("Could not resolve source path")

    @staticmethod
    def _ensure_remote_url(url: str) -> str:
        """Reject anything that is not a well-formed http(s) URL.

        These readers (``pd.read_csv``, ``read_excel``, ``read_parquet``,
        ``read_feather``, ``read_orc``) all accept local filesystem paths and
        ``file://`` URLs, not just http(s) URLs. A resource's ``url`` (or a
        resource view's ``file_url``) is set by dataset editors, not
        sysadmins, so without this check anyone able to edit a resource could
        make the server read and return the contents of an arbitrary local
        file (e.g. ``ckan.ini``, secrets, other users' uploads).
        """
        parsed = urlparse(url)

        if parsed.scheme not in _ALLOWED_URL_SCHEMES or not parsed.netloc:
            raise ValueError(f"Unsupported or unsafe URL: {url!r}. Only http(s) URLs are allowed.")

        return url

    @contextlib.contextmanager
    def _open_source(self) -> Iterator[str]:
        """Yield a local filesystem path for the source.

        A remote http(s) URL is downloaded to a guarded temporary file first
        (timeouts, size cap, SSRF checks — see :mod:`ckanext.tables.net`); a
        file-keeper-backed upload is read via its storage's real path when
        that storage is on local disk, or streamed into a temporary file
        otherwise (e.g. S3, where there's no local path to read from); a
        plain local path (resolved via the legacy CKAN uploader) is yielded
        unchanged. Any temp file created along the way is removed once the
        caller is done.
        """
        source = self.get_source_path()

        if source.startswith(("http://", "https://")):
            with fetch_remote_file(source) as local_path:
                yield local_path
        elif self._upload_storage is not None:
            with self._read_fk_upload(self._upload_storage, source) as local_path:
                yield local_path
        else:
            yield source

    @staticmethod
    @contextlib.contextmanager
    def _read_fk_upload(storage: Any, location: str) -> Iterator[str]:
        """Yield a local path for a file-keeper storage object.

        ``storage.full_path()`` is a cheap, generic ``join(base, location)`` —
        it doesn't confirm the storage is actually on local disk, so it's only
        trusted when the result exists on disk; otherwise (e.g. an S3-backed
        storage) fall back to streaming the content into a temp file.
        """
        # Resolve (and fully exit) this before yielding — a caller's exception while
        # using the yielded path must propagate normally, not get swallowed here.
        local_path = None
        with contextlib.suppress(Exception):
            candidate = storage.full_path(location)
            if os.path.isfile(candidate):
                local_path = candidate

        if local_path is not None:
            yield local_path
            return

        from ckan.lib import files  # noqa: PLC0415 — only importable on CKAN 2.12+

        fd, tmp_path = tempfile.mkstemp(prefix="ckanext-tables-")

        try:
            with os.fdopen(fd, "wb") as tmp_file:
                for chunk in storage.stream(files.FileData(files.Location(location))):
                    tmp_file.write(chunk)
            yield tmp_path
        finally:
            with contextlib.suppress(FileNotFoundError):
                os.remove(tmp_path)

---

DataStoreDataSource

Queries the CKAN Datastore API directly. This source is used automatically when a resource has datastore_active = True. It does not use any caching, as the data is already stored in the database.

from ckanext.tables.shared import DataStoreDataSource

source = DataStoreDataSource(resource_id="<resource-id>")

Filtering, sorting, and pagination are translated into datastore_search parameters:

• = → exact match filter
• like → full-text search (partial word match via PostgreSQL FTS with :*)
• Other comparison operators are not supported by datastore_search and are silently ignored.

Note

DataStoreDataSource requires the datastore plugin to be enabled in CKAN. If it is not active, all methods return empty results gracefully.

A data source that fetches records directly from the CKAN Datastore.

Source code in ckanext/tables/data_sources.py

class DataStoreDataSource(BaseDataSource):
    """A data source that fetches records directly from the CKAN Datastore."""

    def __init__(self, resource_id: str):
        self.resource_id = resource_id

        self._filters: dict[str, Any] = {}
        self._q: dict[str, str] = {}
        self._sort: str | None = None
        self._limit: int | None = None
        self._offset: int | None = None

        self._datastore_enabled = "datastore" in tk.g.plugins

    def filter(self, filters: list[FilterItem]) -> Self:
        self._filters = {}
        self._q = {}

        for filter_item in filters:
            if filter_item.operator == "=":
                self._filters[filter_item.field] = filter_item.value
            elif filter_item.operator == "like":
                # replace non-alphanumeric characters (except dots) with FTS wildcard (_)
                v = str(filter_item.value)
                v = re.sub(r"[^\w\-\.]+", "_", v)
                # append ':*' so we can do partial FTS searches
                self._q[filter_item.field] = v + ":*"
            # Other operators like <, > might require datastore_search_sql
            # which is more complex, so we skip them unless necessary.
        return self

    def sort(self, sort_by: str | None, sort_order: str | None) -> Self:
        self._sort = f"{sort_by} {sort_order or 'asc'}" if sort_by else None
        return self

    def paginate(self, page: int, size: int) -> Self:
        if page and size:
            self._limit = size
            self._offset = (page - 1) * size
        return self

    def all(self) -> list[dict[str, Any]]:
        if not self._datastore_enabled:
            return []

        data_dict: dict[str, Any] = {"resource_id": self.resource_id}

        if self._filters:
            data_dict["filters"] = self._filters

        if self._q:
            data_dict["q"] = json.dumps(self._q)
            data_dict["plain"] = False
            data_dict["language"] = "simple"

        if self._sort:
            data_dict["sort"] = self._sort

        if self._limit is not None:
            data_dict["limit"] = self._limit

        if self._offset is not None:
            data_dict["offset"] = self._offset

        try:
            result = tk.get_action("datastore_search")({}, data_dict)
            return result.get("records", [])
        except (tk.ObjectNotFound, tk.NotAuthorized, tk.ValidationError):
            # ValidationError covers an invalid sort/filter field name — both come
            # straight from unvalidated request params (see filter()/sort() above).
            return []

    def count(self) -> int:
        if not self._datastore_enabled:
            return 0

        data_dict: dict[str, Any] = {"resource_id": self.resource_id, "limit": 0}

        if self._filters:
            data_dict["filters"] = self._filters

        if self._q:
            data_dict["q"] = json.dumps(self._q)
            data_dict["plain"] = False
            data_dict["language"] = "simple"

        try:
            result = tk.get_action("datastore_search")({}, data_dict)
            return result.get("total", 0)
        except (tk.ObjectNotFound, tk.NotAuthorized, tk.ValidationError):
            return 0

    def get_columns(self) -> list[str]:
        if not self._datastore_enabled:
            return []

        data_dict = {"resource_id": self.resource_id, "limit": 0}

        try:
            result = tk.get_action("datastore_search")({}, data_dict)
            return [f["id"] for f in result.get("fields", []) if f["id"] != "_id"]
        except (tk.ObjectNotFound, tk.NotAuthorized):
            return []