async def _create_source_instance(
        cls,
        db: AsyncSession,
        sync: schemas.Sync,
        current_user: schemas.User,
        white_label: Optional[schemas.WhiteLabel] = None,
        access_token: Optional[str] = None,
    ) -> BaseSource:
        """Create and configure the source instance based on authentication type."""
        # Retrieve source connection and model
        source_connection = await crud.connection.get(db, sync.source_connection_id, current_user)
        if not source_connection:
            raise NotFoundException("Source connection not found")

        # Get the source_connection record to access config_fields using sync_id
        source_connection_obj = await crud.source_connection.get_by_sync_id(
            db, sync_id=sync.id, current_user=current_user
        )
        if not source_connection_obj:
            raise NotFoundException("Source connection record not found")

        # Get config fields (will be empty dict if none)
        config_fields = source_connection_obj.config_fields or {}

        source_model = await crud.source.get_by_short_name(db, source_connection.short_name)
        if not source_model:
            raise NotFoundException(f"Source not found: {source_connection.short_name}")

        source_class = resource_locator.get_source(source_model)

        # If access token is provided, use it directly
        if access_token:
            return await source_class.create(access_token, config=config_fields)

        # Otherwise get credentials from database as before
        if not source_connection.integration_credential_id:
            raise NotFoundException("Source connection has no integration credential")

        credential = await cls._get_integration_credential(db, source_connection, current_user)
        decrypted_credential = credentials.decrypt(credential.encrypted_credentials)

        # If the source model requires auth configuration, validate it
        if source_model.auth_config_class:
            auth_config = resource_locator.get_auth_config(source_model.auth_config_class)
            source_credentials = auth_config.model_validate(decrypted_credential)

            # if the source_credential has a refresh token, exchange it for an access token
            if hasattr(source_credentials, "refresh_token") and source_credentials.refresh_token:
                oauth2_response = await oauth2_service.refresh_access_token(
                    db,
                    source_model.short_name,
                    current_user,
                    source_connection.id,
                    decrypted_credential,
                    white_label,
                )
                # Just use the access token
                source_credentials = oauth2_response.access_token
        else:
            source_credentials = decrypted_credential

        # Pass both credentials and config to source creation
        return await source_class.create(source_credentials, config=config_fields)
Create and configure the source instance based on authentication type.
backend/airweave/platform/sync/factory.py