Source code for langchain.utilities.max_compute
from __future__ import annotations
from typing import TYPE_CHECKING, Iterator, List, Optional
from langchain.utils import get_from_env
if TYPE_CHECKING:
    from odps import ODPS
[docs]class MaxComputeAPIWrapper:
    """Interface for querying Alibaba Cloud MaxCompute tables."""
[docs]    def __init__(self, client: ODPS):
        """Initialize MaxCompute document loader.
        Args:
            client: odps.ODPS MaxCompute client object.
        """
        self.client = client
[docs]    @classmethod
    def from_params(
        cls,
        endpoint: str,
        project: str,
        *,
        access_id: Optional[str] = None,
        secret_access_key: Optional[str] = None,
    ) -> MaxComputeAPIWrapper:
        """Convenience constructor that builds the odsp.ODPS MaxCompute client from
            given parameters.
        Args:
            endpoint: MaxCompute endpoint.
            project: A project is a basic organizational unit of MaxCompute, which is
                similar to a database.
            access_id: MaxCompute access ID. Should be passed in directly or set as the
                environment variable `MAX_COMPUTE_ACCESS_ID`.
            secret_access_key: MaxCompute secret access key. Should be passed in
                directly or set as the environment variable
                `MAX_COMPUTE_SECRET_ACCESS_KEY`.
        """
        try:
            from odps import ODPS
        except ImportError as ex:
            raise ImportError(
                "Could not import pyodps python package. "
                "Please install it with `pip install pyodps` or refer to "
                "https://pyodps.readthedocs.io/."
            ) from ex