async def read_sources(
    *,
    db: AsyncSession = Depends(deps.get_db),
    user: schemas.User = Depends(deps.get_user),
) -> list[schemas.Source]:
    """Get all sources with their authentication fields."""
    logger.info("Starting read_sources endpoint")
    try:
        sources = await crud.source.get_all(db)
        logger.info(f"Retrieved {len(sources)} sources from database")
    except Exception as e:
        logger.error(f"Failed to retrieve sources: {str(e)}")
        raise HTTPException(status_code=500, detail="Failed to retrieve sources") from e

    # Initialize auth_fields for each source
    result_sources = []
    invalid_sources = []

    for source in sources:
        try:
            # Strict validation for both config classes
            if not source.auth_config_class:
                invalid_sources.append(f"{source.short_name} (missing auth_config_class)")
                continue

            if not source.config_class:
                invalid_sources.append(f"{source.short_name} (missing config_class)")
                continue

            # Get authentication configuration class
            try:
                auth_config_class = resource_locator.get_auth_config(source.auth_config_class)
                auth_fields = Fields.from_config_class(auth_config_class)
            except AttributeError as e:
                invalid_sources.append(f"{source.short_name} (invalid auth_config_class: {str(e)})")
                continue

            # Get configuration class
            try:
                config_class = resource_locator.get_config(source.config_class)
                config_fields = Fields.from_config_class(config_class)
            except AttributeError as e:
                invalid_sources.append(f"{source.short_name} (invalid config_class: {str(e)})")
                continue

            # Create source model with all fields including auth_fields and config_fields
            source_dict = {
                **{key: getattr(source, key) for key in source.__dict__ if not key.startswith("_")},
                "auth_fields": auth_fields,
                "config_fields": config_fields,
            }

            source_model = schemas.Source.model_validate(source_dict)
            result_sources.append(source_model)

        except Exception as e:
            # Log the error but continue processing other sources
            logger.exception(f"Error processing source {source.short_name}: {str(e)}")
            invalid_sources.append(f"{source.short_name} (error: {str(e)})")

    # Log any invalid sources
    if invalid_sources:
        logger.warning(
            f"Skipped {len(invalid_sources)} invalid sources: {', '.join(invalid_sources)}"
        )

    logger.info(f"Returning {len(result_sources)} valid sources")
    return result_sources
Get all sources with their authentication fields.
backend/airweave/api/v1/endpoints/sources.py