langchain_community.document_loaders.weather

"""简单的读取器，从OpenWeatherMap API读取天气数据"""
from __future__ import annotations

from datetime import datetime
from typing import Iterator, Optional, Sequence

from langchain_core.documents import Document

from langchain_community.document_loaders.base import BaseLoader
from langchain_community.utilities.openweathermap import OpenWeatherMapAPIWrapper


[docs]class WeatherDataLoader(BaseLoader):
    """使用`Open Weather Map` API加载天气数据。

    使用OpenWeatherMap的免费API读取任何位置的预报和当前天气。查看'https://openweathermap.org/appid'了解如何生成免费的OpenWeatherMap API。"""

[docs]    def __init__(
        self,
        client: OpenWeatherMapAPIWrapper,
        places: Sequence[str],
    ) -> None:
        """使用参数进行初始化。"""
        super().__init__()
        self.client = client
        self.places = places

[docs]    @classmethod
    def from_params(
        cls, places: Sequence[str], *, openweathermap_api_key: Optional[str] = None
    ) -> WeatherDataLoader:
        client = OpenWeatherMapAPIWrapper(openweathermap_api_key=openweathermap_api_key)  # type: ignore[call-arg]
        return cls(client, places)

[docs]    def lazy_load(
        self,
    ) -> Iterator[Document]:
        """为给定的位置延迟加载天气数据。"""
        for place in self.places:
            metadata = {"queried_at": datetime.now()}
            content = self.client.run(place)
            yield Document(page_content=content, metadata=metadata)
Source code for langchain_community.document_loaders.weather
