services/utils.py

from urllib.parse import urlparse

def filter_tables(threshold, sorted_results):
    return [doc["content"] for doc in sorted_results if doc["score"] > threshold]

def get_db_scheme_from_uri(uri: str) -> str:
    """
    Given a SQLAlchemy-style connection URI, return its scheme name
    (with any '+driver' suffix stripped).

    Examples:
        >>> get_db_scheme_from_uri("postgresql://user:pass@host/db")
        'postgresql'

        >>> get_db_scheme_from_uri("postgresql+psycopg2://user:pass@host/db")
        'postgresql'

        >>> get_db_scheme_from_uri("duckdb:///path/to/db.duckdb")
        'duckdb'
    """
    parsed = urlparse(uri)
    scheme = parsed.scheme
    if not scheme:
        raise ValueError(f"No scheme found in URI: {uri!r}")
    # Strip any "+driver" suffix (e.g. "mysql+mysqldb")
    return scheme.split("+", 1)[0]

import os
from urllib.parse import urlparse

def extract_filename(path_or_url):
    """
    Extract the file name from a local path or a URL.

    Args:
        path_or_url (str): The file path or URL.

    Returns:
        str: The extracted file name.
    """
    parsed = urlparse(path_or_url)
    if parsed.scheme in ('http', 'https', 'ftp'):
        return os.path.basename(parsed.path)
    else:
        return os.path.basename(path_or_url)