Function _create_segments_from_ld

api/integrations/launch_darkly/services.py:999–1077

Create segments from the given Launch Darkly segments. This also creates inclusion rules for segments. :param ld_segments: A list of mapping from (env, segment). :return A mapping from ld segment key to Segment itself.

(
    import_request: LaunchDarklyImportRequest,
    ld_segments: list[tuple[ld_types.UserSegment, str]],
    environments_by_ld_environment_key: dict[str, Environment],
    tags_by_ld_tag: dict[str, Tag],
    project_id: int,
)

def _create_segments_from_ld(
    import_request: LaunchDarklyImportRequest,
    ld_segments: list[tuple[ld_types.UserSegment, str]],
    environments_by_ld_environment_key: dict[str, Environment],
    tags_by_ld_tag: dict[str, Tag],
    project_id: int,
) -> dict[str, Segment]:
    """
    Create segments from the given Launch Darkly segments. This also creates inclusion rules for segments.

    :param ld_segments: A list of mapping from (env, segment).
    :return A mapping from ld segment key to Segment itself.
    """
    segments_by_ld_key = {}
    for ld_segment, env in ld_segments:
        if ld_segment["deleted"]:
            continue

        # Make sure consecutive updates do not create the same segment.
        segment, _ = Segment.live_objects.update_or_create(
            name=_get_segment_name(ld_segment["name"], env),
            project_id=project_id,
        )

        segments_by_ld_key[ld_segment["key"]] = segment

        # TODO: Tagging segments is not supported yet. https://github.com/Flagsmith/flagsmith/issues/3241

        # Create the segment rule for the segment.
        rules = ld_segment["rules"]
        for rule in rules:
            _create_segment_rule_for_segment(
                import_request=import_request,
                segment=segment,
                clauses=rule["clauses"],
            )

        # Create or update identities that are mentioned in the segment.
        for identifier in ld_segment["included"] + ld_segment["excluded"]:
            identity, _ = Identity.objects.get_or_create(
                identifier=identifier,
                environment=environments_by_ld_environment_key[env],
            )
            identity.update_traits(
                [
                    {
                        "trait_key": "key",
                        "trait_value": identifier,  # type: ignore[typeddict-item]
                    }
                ]
            )

        _include_users_to_segment(
            import_request=import_request,
            segment=segment,
            users=ld_segment["included"],
            negate=False,
        )