Integration guide (Python)

This page gives you a ready-to-use client to call Goodays GEO fragment API from your server. It implements every recommendation of Integration best practices.

What this client does

A developer copies the client into your website's server code. Each time a visitor or a crawler opens the page of a point of sale, the client:

asks Goodays for the fragment of that point of sale;
keeps it in a cache for 24 hours, so your pages don't call Goodays on every visit;
gives up after 2 seconds if Goodays is slow, so your page is never delayed;
if Goodays is unreachable, serves the last version it received, or renders the page without the reviews block.

Your code then only calls one function with the external ID of the point of sale, and inserts the result in the page.

Requirements

Client: goodays_geo.py. It is a single file that only depends on requests.

In the examples below:

madrigal-los-pollos-hermanos is a universe ID (one of your brands): replace it with yours. In the client, it is the client_id parameter.
ABQ-001 is an external ID (your own code for the point of sale): replace it with the code of the point of sale the page is about.

Your Goodays project contact gives you your universe ID and confirms your external IDs.

Use it in your project

Copy goodays_geo.py into your project and add its only dependency (uv add requests, or pip install requests). Then create one client per process. The client holds the cache.

from goodays_geo import GeoFragmentClient

geo = GeoFragmentClient("madrigal-los-pollos-hermanos")

fragment = geo.get_fragment("ABQ-001")  # str, or None: render the page without reviews

Output the fragment unescaped in your server-side template:

{% if fragment %}
  <div id="reviews">{{ fragment|safe }}</div>
{% endif %}

This syntax works the same in Jinja (Flask) and Django templates.

get_fragment() is blocking. In an async framework such as FastAPI or Starlette, declare the route with def or call it with await run_in_threadpool(geo.get_fragment, place).

Options

GeoFragmentClient(
    "madrigal-los-pollos-hermanos",
    base_url="https://geo.goodays.co",
    cache=None,               # any object with get(key) / set(key, value, timeout)
    session=None,             # your own requests.Session (proxies, retries, ...)
    connect_timeout=1.0,
    read_timeout=2.0,
    fresh_ttl=24 * 3600,      # fragments are regenerated every 24 h
    not_found_ttl=3600,
    error_backoff=60,
    stale_max=7 * 24 * 3600,
    live=False,               # development only: live data, no cache
)

Fallbacks are logged on the goodays_geo logger at WARNING level.

Shared cache (several workers)

The default MemoryCache lives inside one process. With gunicorn or uWSGI workers, or with several servers, share the cache so they all reuse the same copy.

Django and Flask-Caching already have the right interface:

from django.core.cache import cache

geo = GeoFragmentClient("madrigal-los-pollos-hermanos", cache=cache)

To use Redis directly:

import json
import redis

class RedisCache:
    def __init__(self, client: redis.Redis) -> None:
        self.client = client

    def get(self, key):
        raw = self.client.get(key)
        return json.loads(raw) if raw else None

    def set(self, key, value, timeout):
        self.client.set(key, json.dumps(value), ex=timeout)

geo = GeoFragmentClient("madrigal-los-pollos-hermanos", cache=RedisCache(redis.Redis.from_url("redis://localhost")))

Client code

Copy this file into your project as goodays_geo.py.

"""Goodays GEO Fragments API client.

Fetches the pre-rendered reviews fragment (HTML + JSON-LD) of a place and keeps
it in a local cache, so rendering your pages never depends on a live call to
Goodays:

- fresh cache hit: no network call at all;
- expired entry: conditional request (ETag / Last-Modified), usually a 304;
- error, timeout or unexpected response: the last good fragment is served
  (stale-on-error) and the origin is not retried before ``error_backoff``;
- unknown place (404) or nothing cached: ``None``, render the page without
  the reviews block.

Copy this file into your project. It only depends on ``requests``.
"""

from __future__ import annotations

import logging
import threading
import time
from typing import Any, Protocol
from urllib.parse import quote

import requests

__all__ = ["DEFAULT_BASE_URL", "GeoFragmentClient", "MemoryCache"]

DEFAULT_BASE_URL = "https://geo.goodays.co"
USER_AGENT = "goodays-geo-snippets/python/1.0"

logger = logging.getLogger("goodays_geo")

class Cache(Protocol):
    """Anything with ``get(key)`` and ``set(key, value, timeout)``.

    Django's cache, Flask-Caching or a thin Redis wrapper all fit.
    """

    def get(self, key: str) -> Any: ...

    def set(self, key: str, value: Any, timeout: int) -> Any: ...

class MemoryCache:
    """Thread-safe in-process cache.

    Good enough for a single process. When you run several workers, use a
    shared cache (Redis, Memcached...) so they all reuse the same fragment.
    """

    def __init__(self, max_entries: int = 1000) -> None:
        self._data: dict[str, tuple[float, Any]] = {}
        self._lock = threading.Lock()
        self._max_entries = max_entries

    def get(self, key: str) -> Any:
        with self._lock:
            item = self._data.get(key)
            if item is None:
                return None
            expires_at, value = item
            if expires_at <= time.time():
                del self._data[key]
                return None
            return value

    def set(self, key: str, value: Any, timeout: int) -> None:
        with self._lock:
            if key not in self._data and len(self._data) >= self._max_entries:
                # Evict the oldest entry (dicts keep insertion order).
                self._data.pop(next(iter(self._data)))
            self._data[key] = (time.time() + timeout, value)

class GeoFragmentClient:
    def __init__(
        self,
        client_id: str,
        *,
        base_url: str = DEFAULT_BASE_URL,
        cache: Cache | None = None,
        session: requests.Session | None = None,
        connect_timeout: float = 1.0,
        read_timeout: float = 2.0,
        fresh_ttl: int = 24 * 3600,  # fragments are regenerated every 24 h
        not_found_ttl: int = 3600,
        error_backoff: int = 60,
        stale_max: int = 7 * 24 * 3600,
        live: bool = False,  # development only: bypass every cache, live data
    ) -> None:
        self.client_id = client_id
        self.base_url = base_url.rstrip("/")
        self.cache = cache if cache is not None else MemoryCache()
        self.session = session or requests.Session()
        self.timeout = (connect_timeout, read_timeout)
        self.fresh_ttl = fresh_ttl
        self.not_found_ttl = not_found_ttl
        self.error_backoff = error_backoff
        self.stale_max = stale_max
        self.live = live
        self._locks: dict[str, threading.Lock] = {}
        self._locks_guard = threading.Lock()

    def fragment_url(self, place: str) -> str:
        return (
            f"{self.base_url}/clients/{quote(self.client_id, safe='')}"
            f"/{quote(place, safe='')}"
            + ("?live=true" if self.live else "")
        )

    def get_fragment(self, place: str) -> str | None:
        """Return the fragment HTML to inject as-is, or ``None``."""
        key = f"goodays-geo:{self.client_id}:{place}"
        entry = self.cache.get(key)
        if not self.live and entry is not None and time.time() < entry["fresh_until"]:
            return entry["html"]

        # One refresh per place at a time: concurrent requests wait for it.
        with self._lock_for(key):
            entry = self.cache.get(key)
            if not self.live and entry is not None and time.time() < entry["fresh_until"]:
                return entry["html"]
            return self._refresh(key, place, entry)

    def _refresh(self, key: str, place: str, entry: dict | None) -> str | None:
        headers = {"Accept": "text/html", "User-Agent": USER_AGENT}
        if entry and entry.get("etag"):
            headers["If-None-Match"] = entry["etag"]
        if entry and entry.get("last_modified"):
            headers["If-Modified-Since"] = entry["last_modified"]

        try:
            response = self.session.get(
                self.fragment_url(place), headers=headers, timeout=self.timeout
            )
        except requests.RequestException as exc:
            return self._fallback(key, entry, f"request failed: {exc}")

        if response.status_code == 304 and entry is not None:
            return self._store(
                key, entry["html"], entry.get("etag"), entry.get("last_modified")
            )

        content_type = response.headers.get("Content-Type", "")
        if response.status_code == 200 and content_type.startswith("text/html"):
            # The fragment is UTF-8; requests would assume ISO-8859-1 when the
            # Content-Type header carries no charset.
            html = response.content.decode("utf-8")
            if html.strip():
                return self._store(
                    key,
                    html,
                    response.headers.get("ETag"),
                    response.headers.get("Last-Modified"),
                )

        if response.status_code == 404:
            logger.info("No GEO fragment for %s/%s (404)", self.client_id, place)
            self._store_miss(key, self.not_found_ttl)
            return None

        return self._fallback(
            key, entry, f"unexpected response: HTTP {response.status_code} {content_type}"
        )

    def _store(
        self, key: str, html: str, etag: str | None, last_modified: str | None
    ) -> str:
        now = time.time()
        entry = {
            "html": html,
            "etag": etag,
            "last_modified": last_modified,
            "fresh_until": now + self.fresh_ttl,
            # Kept that long to be served as a stale copy if Goodays is unreachable.
            "expires_at": now + self.stale_max,
        }
        self.cache.set(key, entry, self.stale_max)
        return html

    def _store_miss(self, key: str, ttl: int) -> None:
        """Remember there is nothing to show, so the origin isn't hit on every page view."""
        now = time.time()
        entry = {
            "html": None,
            "etag": None,
            "last_modified": None,
            "fresh_until": now + ttl,
            "expires_at": now + ttl,
        }
        self.cache.set(key, entry, ttl)

    def _fallback(self, key: str, entry: dict | None, reason: str) -> str | None:
        logger.warning(
            "GEO fragment %s unavailable (%s), serving %s",
            key,
            reason,
            "stale copy" if entry and entry["html"] else "page without reviews",
        )
        # Don't call the origin again before error_backoff, whatever happens.
        if entry is not None and entry["html"] is not None:
            remaining = int(entry["expires_at"] - time.time())
            if remaining > 0:
                entry = {**entry, "fresh_until": time.time() + self.error_backoff}
                self.cache.set(key, entry, remaining)
                return entry["html"]
        self._store_miss(key, self.error_backoff)
        return None

    def _lock_for(self, key: str) -> threading.Lock:
        with self._locks_guard:
            return self._locks.setdefault(key, threading.Lock())