async def create_initial_dag(
        self,
        db: AsyncSession,
        *,
        sync_id: UUID,
        current_user: schemas.User,
        uow: UnitOfWork,
    ) -> schemas.SyncDag:
        """Create an initial DAG with source, entities, and destination.

        Args:
        ----
            db (AsyncSession): The database session.
            sync_id (UUID): The ID of the sync to create the DAG for.
            current_user (schemas.User): The current user.
            uow (UnitOfWork): The unit of work.

        Returns:
        -------
            schemas.SyncDag: The created DAG.
        """
        # Get sync and validate
        sync = await self._get_and_validate_sync(db, sync_id, current_user)

        # Get file chunker transformer
        file_chunker = await self._get_file_chunker(db)

        # Get web fetcher transformer
        web_fetcher = await self._get_web_fetcher(db)

        # Get source connection and entity definitions
        (
            source,
            source_connection,
            entity_definitions,
        ) = await self._get_source_and_entity_definitions(db, sync, current_user)

        # Get or create destinations
        (
            destinations,
            destination_connections,
        ) = await self._get_destinations_and_destination_connections(db, sync, current_user)

        # Initialize DAG components
        nodes: List[DagNodeCreate] = []
        edges: List[DagEdgeCreate] = []
        processed_entity_ids: Set[UUID] = set()

        # Create source and destination nodes
        source_node_id, destination_node_ids = await self._create_source_and_destination_nodes(
            source, source_connection, destinations, destination_connections, nodes
        )

        # Process entity definitions
        await self._process_all_entity_definitions(
            db,
            entity_definitions,
            processed_entity_ids,
            file_chunker,
            web_fetcher,
            source_node_id,
            destination_node_ids,
            nodes,
            edges,
        )

        # Create and return the DAG
        return await self._create_and_save_dag(db, sync, sync_id, nodes, edges, current_user, uow)
Create an initial DAG with source, entities, and destination.

        Args:
        ----
            db (AsyncSession): The database session.
            sync_id (UUID): The ID of the sync to create the DAG for.
            current_user (schemas.User): The current user.
            uow (UnitOfWork): The unit of work.

        Returns:
        -------
            schemas.SyncDag: The created DAG.
        
backend/airweave/core/dag_service.py