async def _get_source_and_entity_definitions(
        self, db: AsyncSession, sync: schemas.Sync, current_user: schemas.User
    ) -> Tuple[schemas.Source, schemas.Connection, Dict]:
        """Get source connection and entity definitions."""
        source_connection = await crud.connection.get(
            db, id=sync.source_connection_id, current_user=current_user
        )
        if not source_connection:
            raise Exception(f"Source connection for {sync.source_connection_id} not found")

        source = await crud.source.get_by_short_name(db, short_name=source_connection.short_name)

        entity_definitions = await crud.entity_definition.get_multi_by_ids(
            db, ids=source.output_entity_definition_ids
        )

        entity_definitions_dict = {
            entity_definition.id: {
                "entity_class": resource_locator.get_entity_definition(entity_definition),
                "entity_definition": entity_definition,
            }
            for entity_definition in entity_definitions
        }

        return source, source_connection, entity_definitions_dict
Get source connection and entity definitions.
backend/airweave/core/dag_service.py