Function process_import_request

api/integrations/launch_darkly/services.py:1106–1185

(
    import_request: LaunchDarklyImportRequest,
)

def process_import_request(
    import_request: LaunchDarklyImportRequest,
) -> None:
    with _complete_import_request(import_request):
        ld_token = _unsign_ld_value(
            import_request.ld_token,
            import_request.created_by.id,
        )
        ld_project_key = import_request.ld_project_key

        ld_client = LaunchDarklyClient(ld_token)

        with closing_stale_connections():
            try:
                ld_environments = ld_client.get_environments(project_key=ld_project_key)
                ld_flags = ld_client.get_flags_by_envs(
                    project_key=ld_project_key,
                    environment_keys=[env["key"] for env in ld_environments],
                )
                ld_flag_tags = ld_client.get_flag_tags()
                # ld_segment_tags = ld_client.get_segment_tags()
                # Keyed by (segment, environment)
                ld_segments: list[tuple[ld_types.UserSegment, str]] = []
                for env in ld_environments:
                    ld_segments_for_env = ld_client.get_segments(
                        project_key=ld_project_key,
                        environment_key=env["key"],
                    )
                    for segment in ld_segments_for_env:
                        ld_segments.append((segment, env["key"]))

            except RequestException as exc:
                _log_error(
                    import_request=import_request,
                    error_message=(
                        f"{exc.__class__.__name__} "
                        f"{str(exc.response.status_code) + ' ' if exc.response else ''}"
                        + f"when requesting {getattr(exc.request, 'path_url', 'unknown')}"
                    ),
                )
                raise

        # Create environments
        environments_by_ld_environment_key = _create_environments_from_ld(
            ld_environments=ld_environments,
            project_id=import_request.project_id,
        )

        # Create segments using `ld_segment_tags`
        # TODO populate with LD tags when https://github.com/Flagsmith/flagsmith/issues/3241 is done
        segment_tags_by_ld_tag: dict[str, Tag] = {}
        segments_by_ld_key = _create_segments_from_ld(
            import_request=import_request,
            ld_segments=ld_segments,
            environments_by_ld_environment_key=environments_by_ld_environment_key,
            tags_by_ld_tag=segment_tags_by_ld_tag,
            project_id=import_request.project_id,
        )