Function map_environment_to_engine

api/util/mappers/engine.py:193–369

Maps Core API's `environments.models.Environment` model instance to the flag_engine environment document. Before building the document, takes care of resolving relationships and feature versions. :param Environment environment: the environment to map :rtype EnvironmentModel

(
    environment: "Environment",
    *,
    with_integrations: bool = True,
)

def map_environment_to_engine(
    environment: "Environment",
    *,
    with_integrations: bool = True,
) -> EnvironmentModel:
    """
    Maps Core API's `environments.models.Environment` model instance to the
    flag_engine environment document.
    Before building the document, takes care of resolving relationships and
    feature versions.

    :param Environment environment: the environment to map
    :rtype EnvironmentModel
    """
    project: "Project" = environment.project
    organisation: "Organisation" = project.organisation

    # Read relationships - grab all the data needed from the ORM here.

    project_segments = [
        ps for ps in project.segments.all() if ps.id == ps.version_of_id
    ]

    project_segment_feature_states_by_segment_id = _get_segment_feature_states(
        project_segments,
        environment.pk,
        latest_environment_feature_version_uuids=(
            {
                efv.uuid
                for efv in EnvironmentFeatureVersion.objects.get_latest_versions_by_environment_id(
                    environment.id
                )
            }
            if environment.use_v2_feature_versioning
            else []
        ),
    )
    # Drop feature-specific segments that have no FeatureSegment in this
    # environment — without one, they have no evaluation path here, and
    # their rules only inflate the environment document.
    project_segments = [
        ps
        for ps in project_segments
        if ps.feature_id is None
        or project_segment_feature_states_by_segment_id.get(ps.pk)
    ]
    project_segment_rules_by_segment_id: Dict[
        int,
        Iterable["SegmentRule"],
    ] = {segment.pk: segment.rules.all() for segment in project_segments}
    environment_feature_states: List["FeatureState"] = _get_prioritised_feature_states(
        [
            feature_state
            for feature_state in environment.feature_states.all()
            if feature_state.feature_segment_id is None
            and feature_state.identity_id is None
        ]
    )