async def run_source_connection(
        self,
        db: AsyncSession,
        source_connection_id: UUID,
        current_user: schemas.User,
        access_token: Optional[str] = None,
    ) -> schemas.SyncJob:
        """Trigger a sync run for a source connection.

        Args:
            db: The database session
            source_connection_id: The ID of the source connection to run
            current_user: The current user
            access_token: Optional access token to use instead of stored credentials

        Returns:
            The created sync job

        Raises:
            HTTPException: If the source connection is not found or has no associated sync
        """
        source_connection = await crud.source_connection.get(
            db=db, id=source_connection_id, current_user=current_user
        )
        if not source_connection:
            raise HTTPException(status_code=404, detail="Source connection not found")

        if not source_connection.sync_id:
            raise HTTPException(status_code=400, detail="Source connection has no associated sync")

        # Trigger the sync run using the sync service
        sync, sync_job, sync_dag = await sync_service.trigger_sync_run(
            db=db, sync_id=source_connection.sync_id, current_user=current_user
        )

        # Store access token directly without validation if provided
        if access_token:
            sync_job.access_token = access_token

        return sync_job
Trigger a sync run for a source connection.

        Args:
            db: The database session
            source_connection_id: The ID of the source connection to run
            current_user: The current user
            access_token: Optional access token to use instead of stored credentials

        Returns:
            The created sync job

        Raises:
            HTTPException: If the source connection is not found or has no associated sync
        
backend/airweave/core/source_connection_service.py