Function _get_entity_maps

sdk/python/feast/utils.py:890–932

(
    registry,
    project,
    feature_views,
)

def _get_entity_maps(
    registry,
    project,
    feature_views,
) -> Tuple[Dict[str, str], Dict[str, ValueType], Set[str]]:
    # TODO(felixwang9817): Support entities that have different types for different feature views.
    entities = registry.list_entities(project, allow_cache=True)

    entity_by_name: Dict[str, "Entity"] = {entity.name: entity for entity in entities}

    entity_name_to_join_key_map: Dict[str, str] = {}
    entity_type_map: Dict[str, ValueType] = {}
    for entity in entities:
        entity_name_to_join_key_map[entity.name] = entity.join_key
    for feature_view in feature_views:
        for entity_name in feature_view.entities:
            entity = entity_by_name.get(entity_name)
            if entity is None:
                from feast.errors import EntityNotFoundException

                raise EntityNotFoundException(entity_name, project=project)
            # User directly uses join_key as the entity reference in the entity_rows for the
            # entity mapping case.
            entity_name = feature_view.projection.join_key_map.get(
                entity.join_key, entity.name
            )
            join_key = feature_view.projection.join_key_map.get(
                entity.join_key, entity.join_key
            )
            entity_name_to_join_key_map[entity_name] = join_key

        for entity_column in feature_view.entity_columns:
            dtype = entity_column.dtype.to_value_type()
            entity_join_key_column_name = feature_view.projection.join_key_map.get(
                entity_column.name, entity_column.name
            )
            entity_type_map[entity_join_key_column_name] = dtype

    return (
        entity_name_to_join_key_map,
        entity_type_map,
        set(entity_name_to_join_key_map.values()),
    )


def _get_table_entity_values(