DB-GPT/dbgpt/datasource/rdbms/conn_hive.py

"""Hive Connector."""
from typing import Any, Optional, cast
from urllib.parse import quote
from urllib.parse import quote_plus as urlquote

from .base import RDBMSConnector


class HiveConnector(RDBMSConnector):
    """Hive connector."""

    db_type: str = "hive"
    """db driver"""
    driver: str = "hive"
    """db dialect"""
    dialect: str = "hive"

    @classmethod
    def from_uri_db(
        cls,
        host: str,
        port: int,
        user: str,
        pwd: str,
        db_name: str,
        engine_args: Optional[dict] = None,
        **kwargs: Any,
    ) -> "HiveConnector":
        """Create a new HiveConnector from host, port, user, pwd, db_name."""
        db_url: str = f"{cls.driver}://{host}:{str(port)}/{db_name}"
        if user and pwd:
            db_url = (
                f"{cls.driver}://{quote(user)}:{urlquote(pwd)}@{host}:{str(port)}/"
                f"{db_name}"
            )
        return cast(HiveConnector, cls.from_uri(db_url, engine_args, **kwargs))

    def table_simple_info(self):
        """Get table simple info."""
        return []

    def get_users(self):
        """Get users."""
        return []

    def get_grants(self):
        """Get grants."""
        return []

    def get_collation(self):
        """Get collation."""
        return "UTF-8"

    def get_charset(self):
        """Get character_set of current database."""
        return "UTF-8"