feat: import rules from LD flags

api/integrations/launch_darkly/client.py

@@ -1,4 +1,4 @@
-from typing import Any, Iterator, Optional
+from typing import Any, Iterator, Optional, TypeVar
 
 from requests import Session
 
@@ -9,6 +9,8 @@
     LAUNCH_DARKLY_API_VERSION,
 )
 
+T = TypeVar("T")
+
 
 class LaunchDarklyClient:
     def __init__(self, token: str) -> None:
@@ -25,7 +27,7 @@ def _get_json_response(
         self,
         endpoint: str,
         params: Optional[dict[str, Any]] = None,
-    ) -> dict[str, Any]:
+    ) -> T:
         full_url = f"{LAUNCH_DARKLY_API_BASE_URL}{endpoint}"
         response = self.client_session.get(full_url, params=params)
         response.raise_for_status()
@@ -34,9 +36,20 @@ def _get_json_response(
     def _iter_paginated_items(
         self,
         collection_endpoint: str,
-        additional_params: Optional[dict[str, str]] = None,
-    ) -> Iterator[dict[str, Any]]:
+        additional_params: Optional[dict[str, Any]] = None,
+        use_legacy_offset_pagination: bool = False,
+    ) -> Iterator[T]:
+        """
+        Iterator over paginated items in the given collection endpoint.
+
+        :param collection_endpoint: endpoint to get the collection of items
+        :param additional_params: Additional parameters to include in the request
+        :param use_legacy_offset_pagination: Whether to use offset based pagination if `next` links do not
+        exist in the response. Some endpoints do not have `next` links and require offset based pagination.
+        :return: Iterator over the items in the collection
+        """
         params = {"limit": LAUNCH_DARKLY_API_ITEM_COUNT_LIMIT_PER_PAGE}
+        offset = 0
         if additional_params:
             params.update(additional_params)
 
@@ -45,7 +58,8 @@ def _iter_paginated_items(
             params=params,
         )
         while True:
-            yield from response_json.get("items") or []
+            items = response_json.get("items") or []
+            yield from items
             links: Optional[dict[str, ld_types.Link]] = response_json.get("_links")
             if (
                 links
@@ -57,6 +71,14 @@ def _iter_paginated_items(
                 response_json = self._get_json_response(
                     endpoint=next_endpoint,
                 )
+            elif use_legacy_offset_pagination and len(items) == params["limit"]:
+                # Offset based pagination
+                offset += params["limit"]
+                params["offset"] = offset
+                response_json = self._get_json_response(
+                    endpoint=collection_endpoint,
+                    params=params,
+                )
             else:
                 return
 
@@ -82,6 +104,9 @@ def get_flags(self, project_key: str) -> list[ld_types.FeatureFlag]:
         return list(
             self._iter_paginated_items(
                 collection_endpoint=endpoint,
+                # Summary should be set to 0 in order to get the full flag data including rules.
+                # https://apidocs.launchdarkly.com/tag/Feature-flags#operation/getFeatureFlags!in=query&path=summary&t=request
+                additional_params={"summary": "0"},
             )
         )
 
@@ -106,3 +131,33 @@ def get_flag_tags(self) -> list[str]:
                 additional_params={"kind": "flag"},
             )
         )
+
+    def get_segment_tags(self) -> list[str]:
+        """operationId: getTags"""
+        endpoint = "/api/v2/tags"
+        return list(
+            self._iter_paginated_items(
+                collection_endpoint=endpoint,
+                additional_params={"kind": "segment"},
+            )
+        )
+
+    def get_segments(
+        self, project_key: str, environment_key: str
+    ) -> list[ld_types.UserSegment]:
+        """operationId: getSegments"""
+        endpoint = f"/api/v2/segments/{project_key}/{environment_key}"
+        return list(
+            self._iter_paginated_items(
+                collection_endpoint=endpoint,
+                additional_params={"limit": 50},
+                use_legacy_offset_pagination=True,
+            )
+        )
+
+    def get_segment(
+        self, project_key: str, environment_key: str, segment_key: str
+    ) -> ld_types.UserSegment:
+        """operationId: getSegment"""
+        endpoint = f"/api/v2/segments/{project_key}/{environment_key}/{segment_key}"
+        return self._get_json_response(endpoint=endpoint)

api/integrations/launch_darkly/models.py

@@ -15,7 +15,7 @@ class LaunchDarklyImportStatus(TypedDict):
     requested_environment_count: int
     requested_flag_count: int
     result: NotRequired[Literal["success", "failure"]]
-    error_message: NotRequired[str]
+    error_messages: list[str]
 
 
 class LaunchDarklyImportRequest(
@@ -44,8 +44,11 @@ def get_update_log_message(self, _) -> Optional[str]:
             return None
         if self.status.get("result") == "success":
             return "LaunchDarkly import completed successfully"
-        if error_message := self.status.get("error_message"):
-            return f"LaunchDarkly import failed with error: {error_message}"
+        if error_messages := self.status.get("error_messages"):
+            if len(error_messages) > 0:
+                return "LaunchDarkly import failed with errors:\n" + "\n".join(
+                    "- " + error_message for error_message in error_messages
+                )
         return "LaunchDarkly import failed"
 
     def get_audit_log_author(self) -> "FFAdminUser":

api/integrations/launch_darkly/serializers.py

@@ -11,7 +11,9 @@ class LaunchDarklyImportRequestStatusSerializer(serializers.Serializer):
         read_only=True,
         allow_null=True,
     )
-    error_message = serializers.CharField(read_only=True, allow_null=True)
+    error_messages = serializers.ListSerializer(
+        child=serializers.CharField(read_only=True)
+    )
 
 
 class CreateLaunchDarklyImportRequestSerializer(serializers.Serializer):