async def get_oauth2_authorization_url(
        self,
        source_short_name: str,
        client_id: Optional[str] = None,
    ) -> schemas.OAuth2AuthUrl:
        """Get the OAuth2 authorization URL for a source.

        Args:
            source_short_name: The short name of the source
            client_id: Optional client ID to override the default one

        Returns:
            The OAuth2 authorization URL

        Raises:
            HTTPException: If the source is not found or doesn't support OAuth2
        """
        # Get the settings for this source to generate the URL
        oauth2_settings = await integration_settings.get_by_short_name(source_short_name)
        if not oauth2_settings:
            raise HTTPException(
                status_code=404, detail=f"Settings not found for source: {source_short_name}"
            )

        if oauth2_settings.auth_type not in [
            AuthType.oauth2,
            AuthType.oauth2_with_refresh,
            AuthType.oauth2_with_refresh_rotating,
        ]:
            raise HTTPException(
                status_code=400,
                detail=f"Source {source_short_name} does not support OAuth2 authentication",
            )

        # Generate the authorization URL
        auth_url = await oauth2_service.generate_auth_url(oauth2_settings, client_id)

        # Return as schema
        return schemas.OAuth2AuthUrl(url=auth_url)
Get the OAuth2 authorization URL for a source.

        Args:
            source_short_name: The short name of the source
            client_id: Optional client ID to override the default one

        Returns:
            The OAuth2 authorization URL

        Raises:
            HTTPException: If the source is not found or doesn't support OAuth2
        
backend/airweave/core/source_connection_service.py