Function fetch_url

AskAgent/python/data_loading/db_load.py:157–200

Fetch content from a URL. Args: url: URL to fetch Returns: Tuple of (content, file_extension)

(url: str)

        return False

async def fetch_url(url: str) -> tuple[str, str | None]:
    """
    Fetch content from a URL.

    Args:
        url: URL to fetch

    Returns:
        Tuple of (content, file_extension)
    """
    print(f"Fetching content from URL: {url}")

    try:
        async with aiohttp.ClientSession() as session, session.get(url) as response:
            if response.status != 200:
                raise ValueError(f"Failed to fetch URL {url}: HTTP {response.status}")

            # Get content type and extension
            content_type = response.headers.get('Content-Type', '').lower()

            # Determine file extension based on content type
            ext = None
            if 'application/json' in content_type:
                ext = '.json'
            elif 'text/csv' in content_type or 'application/csv' in content_type:
                ext = '.csv'
            elif 'application/rss+xml' in content_type or 'application/atom+xml' in content_type or 'text/xml' in content_type:
                ext = '.xml'

            # If extension not determined from content-type, try from URL
            if ext is None:
                path = urlparse(url).path
                if '.' in path:
                    ext = os.path.splitext(path)[1].lower()
                # For URLs ending with 'feed' or similar
                elif any(kw in path.lower() for kw in ['/feed', '/rss', '/podcast']):
                    ext = '.xml'

            # Get content as text
            content = await response.text()
            return content, ext
    except Exception as e:
        print(f"Error fetching URL {url}: {e!s}")
        raise

async def save_url_content(url: str) -> tuple[str, str]:
    """