async def _process_file_entity(
        self,
        db: AsyncSession,
        entity_class: Type[FileEntity],
        entity_definition: schemas.EntityDefinition,
        file_chunker: schemas.Transformer,
        source_node_id: UUID,
        destination_node_id: UUID,
        nodes: List[DagNodeCreate],
        edges: List[DagEdgeCreate],
        processed_entity_ids: Set[UUID],
    ) -> None:
        """Process a file entity and create necessary nodes and edges."""
        # Get parent and chunk entity classes
        parent_entity_class, chunk_entity_class = entity_class.create_parent_chunk_models()

        # Get entity definitions for parent and chunk
        parent_entity_definition = await self._get_entity_definition_for_entity_class(
            db, parent_entity_class
        )
        chunk_entity_definition = await self._get_entity_definition_for_entity_class(
            db, chunk_entity_class
        )

        # Validate that we have the necessary entity definitions
        if not parent_entity_definition or not chunk_entity_definition:
            from airweave.core.logging import logger

            logger.error(f"Missing entity definitions for {entity_class.__name__}")
            if not parent_entity_definition:
                logger.error(f"Missing parent entity definition for {parent_entity_class.__name__}")
            if not chunk_entity_definition:
                logger.error(f"Missing chunk entity definition for {chunk_entity_class.__name__}")
            raise ValueError(f"Missing entity definitions for {entity_class.__name__}")

        # Mark parent and chunk entity definitions as processed
        processed_entity_ids.add(parent_entity_definition.id)
        processed_entity_ids.add(chunk_entity_definition.id)

        # Create file entity node
        file_node_id = uuid4()
        nodes.append(
            DagNodeCreate(
                id=file_node_id,
                type=NodeType.entity,
                name=entity_definition.name,
                entity_definition_id=entity_definition.id,
            )
        )

        # Create file chunker transformer node
        chunker_node_id = uuid4()
        nodes.append(
            DagNodeCreate(
                id=chunker_node_id,
                type=NodeType.transformer,
                name=file_chunker.name,
                transformer_id=file_chunker.id,
            )
        )

        # Create parent chunk node
        parent_node_id = uuid4()
        nodes.append(
            DagNodeCreate(
                id=parent_node_id,
                type=NodeType.entity,
                name=parent_entity_definition.name,
                entity_definition_id=parent_entity_definition.id,
            )
        )

        # Create chunk entity node
        chunk_node_id = uuid4()
        nodes.append(
            DagNodeCreate(  # Default value
                id=chunk_node_id,
                type=NodeType.entity,
                name=chunk_entity_definition.name,
                entity_definition_id=chunk_entity_definition.id,
            )
        )

        # Create edges
        # Source -> File Entity
        edges.append(DagEdgeCreate(from_node_id=source_node_id, to_node_id=file_node_id))

        # File Entity -> File Chunker
        edges.append(DagEdgeCreate(from_node_id=file_node_id, to_node_id=chunker_node_id))

        # File Chunker -> Parent Entity
        edges.append(DagEdgeCreate(from_node_id=chunker_node_id, to_node_id=parent_node_id))

        # File Chunker -> Chunk Entity
        edges.append(DagEdgeCreate(from_node_id=chunker_node_id, to_node_id=chunk_node_id))

        # Parent Entity -> Destination
        edges.append(DagEdgeCreate(from_node_id=parent_node_id, to_node_id=destination_node_id))

        # Chunk Entity -> Destination
        edges.append(DagEdgeCreate(from_node_id=chunk_node_id, to_node_id=destination_node_id))
Process a file entity and create necessary nodes and edges.
backend/airweave/core/dag_service.py