feat: introduce SSE (ref AI-984)

.gitignore

@@ -9,3 +9,7 @@ coverage.xml
 .mypy_cache/
 .ruff_cache/
 .pytest_cache/
+.idea/
+
+# macOS
+.DS_Store

README.md

@@ -4,6 +4,8 @@ The official [Dynatrace](https://dynatrace.com) [OpenFeature](https://openfeatur
 
 The provider evaluates flags in-process: on initialization it fetches your flag configuration from the Dynatrace CDN and refreshes it in the background, so evaluations are local and do not make a network call.
 
+Changes are picked up through a server-sent events (SSE) stream advertised in the flag configuration: a change notification triggers an immediate re-fetch, and polling every 10 minutes acts as a safety net. While the stream is unavailable or disabled, the provider polls every 10 seconds instead.
+
 > **Status:** Early alpha — not yet ready for production use.
 
 ## Installation
@@ -52,6 +54,7 @@ api.set_provider(
 | --- | --- | --- | --- |
 | `config_origin` | `str` | derived from SDK key | Override the flag configuration origin. |
 | `metrics_origin` | `str` | derived from SDK key | Override the evaluation metrics ingest origin. |
+| `disable_sse` | `bool` | `False` | Don't attempt an SSE connection and poll only, every 10 seconds. |
 
 ## Running the e2e tests
 

pyproject.toml

@@ -22,6 +22,8 @@ dependencies = [
     "requests>=2.27.0",
     # 2.3 added HTTPResponse.shutdown, which _cdn uses to enforce the fetch deadline
     "urllib3>=2.3.0",
+    # SSE client (ld_eventsource); the floor is the release _sse was built and verified against
+    "launchdarkly-eventsource>=1.7.3,<2.0.0",
 ]
 requires-python = ">=3.10"
 
@@ -63,6 +65,11 @@ pretty = true
 strict = true
 disallow_any_generics = false
 
+[[tool.mypy.overrides]]
+# ld_eventsource ships no type stubs or py.typed marker
+module = ["ld_eventsource", "ld_eventsource.*"]
+ignore_missing_imports = true
+
 [tool.ruff]
 target-version = "py310"
 exclude = [".git", ".venv", "__pycache__", "venv"]

src/dynatrace/openfeature/_cdn.py

@@ -51,6 +51,11 @@ def is_rate_limited(self) -> bool:
     def is_success(self) -> bool:
         return 200 <= self.status < 300
 
+    @property
+    def last_modified_epoch_s(self) -> float | None:
+        parsed = _parse_http_date(self.last_modified)
+        return parsed.timestamp() if parsed is not None else None
+
 
 @dataclass
 class _CacheState:

src/dynatrace/openfeature/_cdn_config.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+import json
+import typing
+
+#: ``eventStreams[].type`` value identifying the SSE change-notification stream
+STREAM_TYPE_SSE = "sse"
+
+
+def _non_blank(value: typing.Any) -> str | None:
+    return value if isinstance(value, str) and value.strip() else None
+
+
+def extract_sse_url(config_json: str | None, fallback_origin: str | None) -> str | None:
+    """Return the SSE URL of the first ``sse`` entry in ``eventStreams``.
+    A non-blank ``url`` wins; else ``endpoint.requestUri`` on ``endpoint.origin`` or ``fallback_origin``."""
+    if not config_json or not config_json.strip():
+        return None
+    try:
+        config: typing.Any = json.loads(config_json)
+    except ValueError:
+        return None
+    streams = config.get("eventStreams") if isinstance(config, dict) else None
+    if not isinstance(streams, list):
+        return None
+    for stream in streams:
+        if not isinstance(stream, dict) or stream.get("type") != STREAM_TYPE_SSE:
+            continue
+        url = _non_blank(stream.get("url"))
+        if url is not None:
+            return url
+        endpoint = stream.get("endpoint")
+        if not isinstance(endpoint, dict):
+            return None
+        request_uri = _non_blank(endpoint.get("requestUri"))
+        if request_uri is None:
+            return None
+        origin = _non_blank(endpoint.get("origin")) or _non_blank(fallback_origin)
+        return origin + request_uri if origin is not None else None
+    return None