feat: Launchdarkly importer

api/Makefile

@@ -84,3 +84,17 @@ django-collect-static:
 .PHONY: serve
 serve:
 	poetry run gunicorn --bind 0.0.0.0:8000 app.wsgi --reload
+
+.PHONY: generate-ld-client-types
+generate-ld-client-types:
+	curl -sSL https://app.launchdarkly.com/api/v2/openapi.json | \
+	npx openapi-format /dev/fd/0 \
+		--filterFile ld-openapi-filter.yaml | \
+	datamodel-codegen \
+		--output integrations/launch_darkly/types.py \
+		--output-model-type typing.TypedDict \
+		--target-python-version 3.10 \
+		--use-double-quotes \
+		--use-standard-collections \
+		--wrap-string-literal \
+		--special-field-name-prefix=

api/app/settings/common.py

@@ -148,6 +148,7 @@
     "integrations.webhook",
     "integrations.dynatrace",
     "integrations.flagsmith",
+    "integrations.launch_darkly",
     # Rate limiting admin endpoints
     "axes",
     "telemetry",

api/audit/related_object_type.py

@@ -8,3 +8,4 @@ class RelatedObjectType(enum.Enum):
     ENVIRONMENT = "Environment"
     CHANGE_REQUEST = "Change request"
     EDGE_IDENTITY = "Edge Identity"
+    IMPORT_REQUEST = "Import request"

api/integrations/launch_darkly/apps.py

@@ -0,0 +1,6 @@
+# -*- coding: utf-8 -*-
+from django.apps import AppConfig
+
+
+class LaunchDarklyConfigurationConfig(AppConfig):
+    name = "integrations.launch_darkly"

api/integrations/launch_darkly/client.py

@@ -0,0 +1,108 @@
+from typing import Any, Iterator, Optional
+
+from requests import Session
+
+from integrations.launch_darkly import types as ld_types
+from integrations.launch_darkly.constants import (
+    LAUNCH_DARKLY_API_BASE_URL,
+    LAUNCH_DARKLY_API_ITEM_COUNT_LIMIT_PER_PAGE,
+    LAUNCH_DARKLY_API_VERSION,
+)
+
+
+class LaunchDarklyClient:
+    def __init__(self, token: str) -> None:
+        client_session = Session()
+        client_session.headers.update(
+            {
+                "Authorization": token,
+                "LD-API-Version": LAUNCH_DARKLY_API_VERSION,
+            }
+        )
+        self.client_session = client_session
+
+    def _get_json_response(
+        self,
+        endpoint: str,
+        params: Optional[dict[str, Any]] = None,
+    ) -> dict[str, Any]:
+        full_url = f"{LAUNCH_DARKLY_API_BASE_URL}{endpoint}"
+        response = self.client_session.get(full_url, params=params)
+        response.raise_for_status()
+        return response.json()
+
+    def _iter_paginated_items(
+        self,
+        collection_endpoint: str,
+        additional_params: Optional[dict[str, str]] = None,
+    ) -> Iterator[dict[str, Any]]:
+        params = {"limit": LAUNCH_DARKLY_API_ITEM_COUNT_LIMIT_PER_PAGE}
+        if additional_params:
+            params.update(additional_params)
+
+        response_json = self._get_json_response(
+            endpoint=collection_endpoint,
+            params=params,
+        )
+        while True:
+            yield from response_json.get("items") or []
+            links: Optional[dict[str, ld_types.Link]] = response_json.get("_links")
+            if (
+                links
+                and (next_link := links.get("next"))
+                and (next_endpoint := next_link.get("href"))
+            ):
+                # Don't specify params here because links.next.href includes the
+                # original limit and calculates offsets accordingly.
+                response_json = self._get_json_response(
+                    endpoint=next_endpoint,
+                )
+            else:
+                return
+
+    def get_project(self, project_key: str) -> ld_types.Project:
+        """operationId: getProject"""
+        endpoint = f"/api/v2/projects/{project_key}"
+        return self._get_json_response(
+            endpoint=endpoint, params={"expand": "environments"}
+        )
+
+    def get_environments(self, project_key: str) -> list[ld_types.Environment]:
+        """operationId: getEnvironmentsByProject"""
+        endpoint = f"/api/v2/projects/{project_key}/environments"
+        return list(
+            self._iter_paginated_items(
+                collection_endpoint=endpoint,
+            )
+        )
+
+    def get_flags(self, project_key: str) -> list[ld_types.FeatureFlag]:
+        """operationId: getFeatureFlags"""
+        endpoint = f"/api/v2/flags/{project_key}"
+        return list(
+            self._iter_paginated_items(
+                collection_endpoint=endpoint,
+            )
+        )
+
+    def get_flag_count(self, project_key: str) -> int:
+        """operationId: getFeatureFlags
+
+        Request minimal info and return the total flag count.
+        """
+        endpoint = f"/api/v2/flags/{project_key}"
+        flags: ld_types.FeatureFlags = self._get_json_response(
+            endpoint=endpoint,
+            params={"limit": 1},
+        )
+        return flags["totalCount"]
+
+    def get_flag_tags(self) -> list[str]:
+        """operationId: getTags"""
+        endpoint = "/api/v2/tags"
+        return list(
+            self._iter_paginated_items(
+                collection_endpoint=endpoint,
+                additional_params={"kind": "flag"},
+            )
+        )

api/integrations/launch_darkly/constants.py

@@ -0,0 +1,7 @@
+LAUNCH_DARKLY_API_BASE_URL = "https://app.launchdarkly.com"
+LAUNCH_DARKLY_API_VERSION = "20220603"
+# Maximum limit for /api/v2/projects/
+# /api/v2/flags/ seemingly not limited, but let's not get too greedy
+LAUNCH_DARKLY_API_ITEM_COUNT_LIMIT_PER_PAGE = 1000
+
+LAUNCH_DARKLY_IMPORTED_TAG_COLOR = "#3d4db6"

api/integrations/launch_darkly/migrations/0001_initial.py

@@ -0,0 +1,129 @@
+# Generated by Django 3.2.20 on 2023-09-17 14:34
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+import simple_history.models
+
+
+class Migration(migrations.Migration):
+    initial = True
+
+    dependencies = [
+        ("api_keys", "0003_masterapikey_is_admin"),
+        ("projects", "0019_add_limits"),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="LaunchDarklyImportRequest",
+            fields=[
+                (
+                    "id",
+                    models.AutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("updated_at", models.DateTimeField(auto_now=True)),
+                ("completed_at", models.DateTimeField(blank=True, null=True)),
+                ("ld_project_key", models.CharField(max_length=2000)),
+                ("ld_token", models.CharField(max_length=2000)),
+                ("status", models.JSONField()),
+                (
+                    "created_by",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        to=settings.AUTH_USER_MODEL,
+                    ),
+                ),
+                (
+                    "project",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        to="projects.project",
+                    ),
+                ),
+            ],
+            options={
+                "abstract": False,
+            },
+        ),
+        migrations.CreateModel(
+            name="HistoricalLaunchDarklyImportRequest",
+            fields=[
+                (
+                    "id",
+                    models.IntegerField(
+                        auto_created=True, blank=True, db_index=True, verbose_name="ID"
+                    ),
+                ),
+                ("created_at", models.DateTimeField(blank=True, editable=False)),
+                ("updated_at", models.DateTimeField(blank=True, editable=False)),
+                ("completed_at", models.DateTimeField(blank=True, null=True)),
+                ("ld_project_key", models.CharField(max_length=2000)),
+                ("ld_token", models.CharField(max_length=2000)),
+                ("status", models.JSONField()),
+                ("history_id", models.AutoField(primary_key=True, serialize=False)),
+                ("history_date", models.DateTimeField()),
+                ("history_change_reason", models.CharField(max_length=100, null=True)),
+                (
+                    "history_type",
+                    models.CharField(
+                        choices=[("+", "Created"), ("~", "Changed"), ("-", "Deleted")],
+                        max_length=1,
+                    ),
+                ),
+                (
+                    "created_by",
+                    models.ForeignKey(
+                        blank=True,
+                        db_constraint=False,
+                        null=True,
+                        on_delete=django.db.models.deletion.DO_NOTHING,
+                        related_name="+",
+                        to=settings.AUTH_USER_MODEL,
+                    ),
+                ),
+                (
+                    "history_user",
+                    models.ForeignKey(
+                        null=True,
+                        on_delete=django.db.models.deletion.SET_NULL,
+                        related_name="+",
+                        to=settings.AUTH_USER_MODEL,
+                    ),
+                ),
+                (
+                    "master_api_key",
+                    models.ForeignKey(
+                        blank=True,
+                        null=True,
+                        on_delete=django.db.models.deletion.DO_NOTHING,
+                        to="api_keys.masterapikey",
+                    ),
+                ),
+                (
+                    "project",
+                    models.ForeignKey(
+                        blank=True,
+                        db_constraint=False,
+                        null=True,
+                        on_delete=django.db.models.deletion.DO_NOTHING,
+                        related_name="+",
+                        to="projects.project",
+                    ),
+                ),
+            ],
+            options={
+                "verbose_name": "historical launch darkly import request",
+                "ordering": ("-history_date", "-history_id"),
+                "get_latest_by": "history_date",
+            },
+            bases=(simple_history.models.HistoricalChanges, models.Model),
+        ),
+    ]

api/integrations/launch_darkly/migrations/0002_importrequest_unique_project_ld_project_key_status_result_null.py

@@ -0,0 +1,20 @@
+# Generated by Django 3.2.20 on 2023-10-03 17:30
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("launch_darkly", "0001_initial"),
+    ]
+
+    operations = [
+        migrations.AddConstraint(
+            model_name="launchdarklyimportrequest",
+            constraint=models.UniqueConstraint(
+                condition=models.Q(("status__result__isnull", True)),
+                fields=("project", "ld_project_key"),
+                name="unique_project_ld_project_key_status_result_null",
+            ),
+        ),
+    ]

api/integrations/launch_darkly/models.py

@@ -0,0 +1,64 @@
+from typing import TYPE_CHECKING, Literal, Optional, TypedDict
+
+from core.models import abstract_base_auditable_model_factory
+from django.db import models
+from typing_extensions import NotRequired
+
+from audit.related_object_type import RelatedObjectType
+from projects.models import Project
+
+if TYPE_CHECKING:  # pragma: no cover
+    from users.models import FFAdminUser
+
+
+class LaunchDarklyImportStatus(TypedDict):
+    requested_environment_count: int
+    requested_flag_count: int
+    result: NotRequired[Literal["success", "failure"]]
+    error_message: NotRequired[str]
+
+
+class LaunchDarklyImportRequest(
+    abstract_base_auditable_model_factory(),
+):
+    history_record_class_path = "features.models.HistoricalLaunchDarklyImportRequest"
+    related_object_type = RelatedObjectType.IMPORT_REQUEST
+
+    created_by = models.ForeignKey("users.FFAdminUser", on_delete=models.CASCADE)
+    project = models.ForeignKey(Project, on_delete=models.CASCADE)
+
+    created_at = models.DateTimeField(auto_now_add=True)
+    updated_at = models.DateTimeField(auto_now=True)
+    completed_at = models.DateTimeField(null=True, blank=True)
+
+    ld_project_key = models.CharField(max_length=2000)
+    ld_token = models.CharField(max_length=2000)
+
+    status: LaunchDarklyImportStatus = models.JSONField()
+
+    def get_create_log_message(self, _) -> str:
+        return "New LaunchDarkly import requested"
+
+    def get_update_log_message(self, _) -> Optional[str]:
+        if not self.completed_at:
+            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}"
+        return "LaunchDarkly import failed"
+
+    def get_audit_log_author(self) -> "FFAdminUser":
+        return self.created_by
+
+    def _get_project(self) -> Project:
+        return self.project
+
+    class Meta:
+        constraints = [
+            models.UniqueConstraint(
+                name="unique_project_ld_project_key_status_result_null",
+                fields=["project", "ld_project_key"],
+                condition=models.Q(status__result__isnull=True),
+            )
+        ]