Python Firebase RemoteConfig Client

remote_config.py

from typing import Any

import requests
from firebase_admin import App

ValueType = str | bool | int | float | dict | list


class RemoteConfigClient:
    def __init__(self, app: App):
        self.app = app
        self.access_token = self._get_access_token()
        self.project_id = self.app.project_id
        self.base_url = f"https://firebaseremoteconfig.googleapis.com/v1/projects/{self.project_id}/remoteConfig"
        self._template = None
        self._e_tag = None

    @property
    def config(self):
        return self.fetch_and_activate()

    def _get_access_token(self) -> str:
        return self.app.credential.get_access_token().access_token

    def _get_headers(
        self,
        include_e_tag: bool = False,
        force_update: bool = False,
    ) -> dict:
        """Get headers for API requests.

        Args:
            include_e_tag: Whether to include ETag header for optimistic concurrency control
            force_update: If True, uses If-Match: * to force update regardless of version
        """
        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Accept": "application/json",
            "Content-Type": "application/json; UTF-8",
            "x-goog-user-project": self.project_id,
        }
        if include_e_tag:
            headers["If-Match"] = "*" if force_update else self._e_tag
        return headers

    def fetch_and_activate(self, version_number: str | None = None):
        """Get a project's Remote Config template and associated ETag header.

        Args:
            version_number: Optional version number of the template to look up.
                          If not specified, the latest template will be returned.
        """
        params = {"versionNumber": version_number} if version_number else None
        response = requests.get(
            self.base_url,
            headers=self._get_headers(),
            params=params,
        )
        try:
            response.raise_for_status()
        except requests.exceptions.HTTPError:
            print(f"Error response: {response.text}")
            raise
        self._template = response.json()
        self._e_tag = response.headers.get("ETag")
        return self._template

    def _get_parameter(self, key: str, default_value: Any = None) -> ValueType:
        if not self._template:
            self._template = self.fetch_and_activate()

        parameters = self._template.setdefault("parameters", {})
        return parameters.get(key, {}).get("defaultValue", {}).get("value", default_value)

    def get_string(self, key: str, default="") -> str:
        return str(self._get_parameter(key, default))

    def get_bool(self, key: str, default=False) -> bool:
        value = self._get_parameter(key, default)
        if isinstance(value, bool):
            return value
        return str(value).lower() == "true"

    def get_int(self, key: str, default=0) -> int:
        """Gets an integer value from Remote Config."""
        try:
            return int(self._get_parameter(key, default))
        except (ValueError, TypeError):
            return default

    def set_value(
        self,
        key: str,
        value: ValueType,
        description: str | None = None,
        force_update: bool = False,
    ):
        """Sets a value in Remote Config.

        Args:
            key: The parameter key
            value: The value to set. Can be:
                - string
                - boolean ("true" or "false")
                - number (integer or float)
                - JSON (dict or list)
            description: Optional description for the parameter
            force_update: If True, forces update regardless of current version
        """
        if not self._template:
            self._template = self.fetch_and_activate()

        # Determine value type based on Python type
        if isinstance(value, bool):
            value_type = "BOOLEAN"
        elif isinstance(value, (int, float)):
            value_type = "NUMBER"
        elif isinstance(value, (dict, list)):
            value_type = "JSON"
        else:
            value_type = "STRING"

        # Construct parameter with proper value type
        parameter = {
            "defaultValue": {
                "value": str(value),  # Values must be strings in the API
            },
            "valueType": value_type,  # valueType at parameter level
        }
        if description:
            parameter["description"] = description

        # Update template
        if "parameters" not in self._template:
            self._template["parameters"] = {}
        self._template["parameters"][key] = parameter

        # Use ETag in headers to prevent conflicts
        response = requests.put(
            self.base_url,
            headers=self._get_headers(include_e_tag=True, force_update=force_update),
            json=self._template,
        )
        try:
            response.raise_for_status()
        except requests.exceptions.HTTPError:
            print(f"Error response: {response.text}")
            raise

        # Update stored template and ETag
        self._template = response.json()
        self._e_tag = response.headers.get("ETag")
        print(f"Set {key} to {value}")

    def list_versions(
        self,
        page_size: int | None = None,
        page_token: str | None = None,
    ):
        """Get a list of Remote Config template versions.

        Only the last 300 versions are stored.
        """
        params = {}
        if page_size:
            params["pageSize"] = page_size
        if page_token:
            params["pageToken"] = page_token

        response = requests.get(
            f"{self.base_url}:listVersions",
            headers=self._get_headers(),
            params=params,
        )
        response.raise_for_status()
        return response.json()

    def rollback(self, version_number: str):
        """Roll back to a specific version.

        Args:
            version_number: The version number to roll back to
        """
        response = requests.post(
            f"{self.base_url}:rollback",
            headers=self._get_headers(),
            json={"versionNumber": version_number},
        )
        response.raise_for_status()
        self._template = response.json()
        self._e_tag = response.headers.get("ETag")
        return self._template