owid-catalog: Data APIs¶

def search(
    self,
    query: str,
    *,
    countries: list[str] | None = None,
    topics: list[str] | None = None,
    require_all_countries: bool = False,
    limit: int = 10,
    page: int = 0,
    timeout: int | None = None,
) -> ResponseSet[ChartResult]:
    """Search for charts matching a query.

    Args:
        query: Search query string.
        countries: Optional list of country names to filter by.
        topics: Optional list of topic names to filter by.
        require_all_countries: If True, only return charts with ALL
            specified countries. Default False (any country matches).
        limit: Maximum results to return (1-100). Default 20.
        page: Page number for pagination (0-indexed). Default 0.
        timeout: HTTP request timeout in seconds. Defaults to client timeout.

    Returns:
        ResponseSet containing ChartResult objects, sorted by popularity (most viewed first).
        Each result includes a `popularity` field (0.0-1.0) based on analytics views.

    Example:
        ```python
        # Basic search (sorted by popularity)
        results = client.charts.search("life expectancy")
        for chart in results:
            print(f"{chart.title}: popularity={chart.popularity:.3f}")

        # Filter by countries
        results = client.charts.search(
            "gdp",
            countries=["France", "Germany"],
            require_all_countries=True
        )

        # Get data from search results
        tb = results[0].fetch()
        ```
    """
    return self._client._site_search.charts(
        query=query,
        countries=countries,
        topics=topics,
        require_all_countries=require_all_countries,
        limit=limit,
        page=page,
        timeout=timeout or self._client.timeout,
    )

def fetch(
    self,
    slug_or_url: str,
    *,
    type: ChartType | None = None,
    load_data: bool = True,
    timeout: int | None = None,
) -> ChartTable:
    """Fetch chart data as a ChartTable with rich metadata.

    Accepts a chart slug, a slug with query parameters, or a full URL. The slug,
    query parameters, and chart type are extracted automatically.

    Args:
        slug_or_url: One of:

            - Chart slug: ``"life-expectancy"``
            - Slug with query params: ``"education-spending?level=primary&spending_type=gdp_share"``
            - Full grapher URL: ``"https://ourworldindata.org/grapher/life-expectancy?tab=table"``
            - Full explorer URL: ``"https://ourworldindata.org/explorers/covid?Metric=Cases"``
        type: Override the chart type. Defaults to ``"chart"`` (grapher).
            Use ``"explorerView"`` for explorer views. Auto-detected from full URLs.
        load_data: If True (default), load full chart data.
                   If False, load only structure (columns and metadata) without rows.
        timeout: HTTP request timeout in seconds. Defaults to client timeout.

    Returns:
        ChartTable with chart data and chart_config. Column metadata (unit, description, etc.)
        is populated from the chart's metadata.json. Chart config is accessible via .metadata.chart_config.

    Note:
        Explorer views are best-effort. Some explorers may return 503 or other errors
        from their CSV endpoint.

    Example:
        ```python
        # Fetch a grapher chart by slug
        tb = client.charts.fetch("life-expectancy")

        # Fetch with query params (e.g., a multiDim view)
        tb = client.charts.fetch("education-spending?level=primary&spending_type=gdp_share")

        # Fetch from a full URL (type and query params auto-detected)
        tb = client.charts.fetch("https://ourworldindata.org/explorers/covid?Metric=Cases")

        # Explicitly fetch an explorer view
        tb = client.charts.fetch("covid?Metric=Cases", type="explorerView")
        ```
    """
    effective_timeout = timeout or self._client.timeout
    parsed = parse_chart_slug(slug_or_url)

    slug = parsed.slug
    effective_type = type or parsed.type
    effective_params = parsed.query_params

    # Pick base URL based on type
    if effective_type == EXPLORER:
        base_url = f"{self._site_url}/explorers"
        other_type: ChartType = CHART
    else:
        base_url = self.base_url
        other_type = EXPLORER

    try:
        return _load_chart_table(
            slug,
            load_data=load_data,
            timeout=effective_timeout,
            use_column_short_names=True,
            base_url=base_url,
            extra_params=_parse_query_params(effective_params) if effective_params else None,
        )
    except (requests.HTTPError, ChartNotFoundError) as e:
        hint = "an explorer view" if other_type == EXPLORER else "a grapher chart"
        raise e.__class__(f'{e}\n\nIf this is {hint}, try: fetch("{slug_or_url}", type={other_type!r})') from e

def search(
    self,
    table: str | None = None,
    namespace: str | None = None,
    version: str | None = None,
    dataset: str | None = None,
    channel: str | None = None,
    case: bool = False,
    match: Literal["exact", "contains", "regex", "fuzzy"] = "exact",
    fuzzy_threshold: int = 70,
    timeout: int | None = None,
    refresh_index: bool = False,
    latest: bool = False,
) -> ResponseSet[TableResult]:
    """Search the catalog for tables matching criteria.

    Args:
        table: Table name pattern to search for
        namespace: Filter by namespace (exact match)
        version: Filter by version (exact match)
        dataset: Dataset name pattern to search for
        channel: Filter by channel (exact match). Defaults to 'garden' if not specified.
        case: Case-sensitive search (default: False)
        match: How to match table/dataset names (default: "exact"):
            - "fuzzy": Typo-tolerant similarity matching
            - "exact": Exact string match
            - "contains": Substring match
            - "regex": Regular expression pattern
        fuzzy_threshold: Minimum similarity score 0-100 for fuzzy matching.
            Only used when match="fuzzy". (default: 70)
        timeout: HTTP request timeout in seconds for catalog loading. Defaults to client timeout.
        refresh_index: If True, force re-download of the catalog index. Default False.
        latest: If True, keep only the latest version of each table
            (grouped by namespace, dataset, table, channel). Default False.
            Note: results without a version are dropped when this is enabled.

    Returns:
        ResponseSet containing matching TableResult objects, sorted by popularity (most viewed first).
        If match="fuzzy", results are sorted by fuzzy relevance score instead.
        Each result includes a `popularity` field (0.0-1.0) based on analytics views.

    Example:
        ```python
        # Exact match (default) - searches garden channel by default
        results = client.tables.search(table="population")

        # Substring match
        results = client.tables.search(table="pop", match="contains")

        # Regex search
        results = client.tables.search(table="population.*density", match="regex")

        # Fuzzy search sorted by relevance
        results = client.tables.search(table="populaton", match="fuzzy")

        # Case-sensitive fuzzy search with custom threshold
        results = client.tables.search(table="GDP", match="fuzzy", case=True, fuzzy_threshold=85)

        # Filter by namespace and version
        results = client.tables.search(
            table="wdi",
            namespace="worldbank_wdi",
            version="2025-09-08",
        )

        # Search in a specific channel
        results = client.tables.search(
            table="wdi",
            namespace="worldbank_wdi",
            version="2025-09-08",
            channel="meadow",
        )

        # Load a specific result
        tb = results[0].fetch()
        ```
    """
    # Default to garden channel if not specified
    if channel is None:
        channel = "garden"

    # Load and filter catalog index
    index = self._get_index(timeout=timeout, force=refresh_index)
    matches, _ = self._filter_index(
        index,
        table=table,
        dataset=dataset,
        namespace=namespace,
        version=version,
        channel=channel,
        match=match,
        case=case,
        fuzzy_threshold=fuzzy_threshold,
    )

    # Fetch popularity data
    popularity = self._fetch_popularity(matches, timeout)

    # Convert to results
    results = self._to_results(matches, popularity)

    # Keep only latest version per group if requested
    if latest:
        results = _keep_latest_versions(
            results,
            key=lambda r: (r.namespace, r.dataset, r.table, r.channel),
        )

    # Build descriptive query from search parameters
    query = self._build_query(
        table=table,
        namespace=namespace,
        version=version,
        dataset=dataset,
        channel=channel,
    )

    return ResponseSet(
        items=results,
        query=query,
        total_count=len(results),
        base_url=self.catalog_url,
        _ui_advanced=False,
    )

def fetch(
    self,
    path: str,
    *,
    load_data: bool = True,
    formats: list[str] | None = None,
    is_public: bool = True,
    timeout: int | None = None,
) -> Table:
    """Fetch a table by catalog path.

    Loads the table directly from the catalog.

    Args:
        path: Full catalog path (e.g., "garden/un/2024-07-12/un_wpp/population").
        load_data: If True (default), load full table data.
                   If False, load only table structure (columns and metadata) without rows.
        formats: List of formats to try. If None, tries all supported formats.
        is_public: Whether the table is publicly accessible. Default True.
        timeout: HTTP request timeout in seconds (currently unused, reserved for future).

    Returns:
        Table with data and metadata (or just metadata if load_data=False).

    Raises:
        ValueError: If path format is invalid.
        KeyError: If table not found at path.

    Example:
        ```python
        # Load table with data
        tb = client.tables.fetch("garden/un/2024-07-12/un_wpp/population")
        print(tb.head())

        # Load only metadata (no data rows)
        tb = client.tables.fetch("garden/un/2024-07-12/un_wpp/population", load_data=False)
        print(tb.columns)
        ```
    """
    # Validate path format
    catalog_path = CatalogPath.from_str(path)

    if not catalog_path.table:
        raise ValueError(f"Invalid path format: {path}. Expected format: channel/namespace/version/dataset/table")

    return _load_table(
        path, formats=formats, is_public=is_public, load_data=load_data, catalog_url=self.catalog_url
    )

def search(
    self,
    query: str,
    *,
    limit: int = 10,
    show_legacy: bool = False,
    latest: bool = False,
    sort_by: Literal["relevance", "similarity"] = "relevance",
    timeout: int | None = None,
) -> ResponseSet[IndicatorResult]:
    """Search for indicators using natural language.

    Uses semantic search to find indicators that match the
    meaning of your query, not just keyword matching.

    Args:
        query: Natural language search query
            (e.g., "renewable energy capacity", "child mortality rate").
        limit: Maximum number of results to return. Default 10.
        show_legacy: If True, show pre-ETL indicators only. Default False.
        latest: If True, keep only the latest version of each indicator
            (grouped by namespace, dataset, column_name). Default False.
            Note: results without a version are dropped when this is enabled.
        sort_by: How to sort results (default: "relevance"):

            - "relevance": Combined score blending semantic similarity (60%) and popularity (40%).
            - "similarity": Sort by semantic similarity score only.
        timeout: HTTP request timeout in seconds. Defaults to client timeout.

    Returns:
        SearchResults containing IndicatorResult objects, sorted according to `sort_by`.
        Each result includes a `popularity` field (0.0-1.0) based on analytics views.

    Example:
        ```python
        # Search for indicators (sorted by relevance by default)
        results = client.indicators.search("CO2 emissions per capita")

        # View results
        for ind in results:
            print(f"{ind.title}")
            print(f"  Score: {ind.score:.3f}")
            print(f"  Popularity: {ind.popularity:.3f}")

        # Load data from top result
        tb = results[0].fetch()

        # Sort by semantic similarity only (original behavior)
        results = client.indicators.search("CO2 emissions", sort_by="similarity")
        ```
    """
    params = {
        "query": query,
        "limit": limit,
    }

    resp = session.get(self.search_url, params=params, timeout=timeout or self._client.timeout)

    # Handle HTTP errors with informative messages from response body
    if not resp.ok:
        error_detail = ""
        try:
            error_data = resp.json()
            error_detail = error_data.get("detail", "")
        except Exception:
            pass

        if error_detail:
            raise requests.HTTPError(
                f"{resp.status_code} Error for url: {resp.url} - {error_detail}",
                response=resp,
            )
        else:
            resp.raise_for_status()

    data = resp.json()

    raw_results: list[tuple[dict[str, Any], str | None]] = []
    for r in data.get("results", []):
        path = r.get("catalog_path", "")

        # If legacy indicator, keep only if asked to
        if "/NULL/" in path:
            if not show_legacy:
                # Skip legacy indicators unless requested
                continue
            path = None
        raw_results.append((r, path))

    # Fetch popularity via Datasette API
    slugs = [path for _, path in raw_results if path]
    popularity_data = (
        self._client._datasette.fetch_popularity(
            slugs,
            type="indicator",
            timeout=timeout or self._client.timeout,
        )
        if slugs
        else {}
    )

    results = [
        IndicatorResult(
            indicator_id=r.get("indicator_id", 0),
            title=r.get("title", ""),
            score=r.get("score", 0.0),
            path=path,
            description=r.get("description", ""),
            column_name=r.get("metadata", {}).get("column", ""),
            unit=r.get("metadata", {}).get("unit", ""),
            n_charts=r.get("n_charts", 0),
            popularity=popularity_data.get(path, 0.0) if path else 0.0,
            catalog_url=self.catalog_url,
        )
        for r, path in raw_results
    ]

    # Drop indicators without a version (unless show_legacy=True)
    if not show_legacy:
        results = [r for r in results if r.version is not None]

    # Keep only latest version per group if requested
    if latest:
        results = _keep_latest_versions(
            results,
            key=lambda r: (r.namespace, r.dataset, r.column_name),
        )

    # Sort results according to sort_by parameter
    if sort_by == "similarity":
        # Original behavior: sort by semantic score only
        results.sort(key=lambda r: r.score, reverse=True)
    else:
        # "relevance" (default): blend similarity and popularity
        # Normalize scores to 0-1 range for blending
        max_score = max((r.score for r in results), default=1.0) or 1.0
        results.sort(
            key=lambda r: 0.6 * (r.score / max_score) + 0.4 * r.popularity,
            reverse=True,
        )

    return ResponseSet(
        items=results,
        query=query,
        total_count=data.get("total_results", len(results)),
        base_url=self.catalog_url,
        _ui_advanced=False,
    )