Source code for langchain_community.document_loaders.arcgis_loader

"""Document Loader for ArcGIS FeatureLayers."""

from __future__ import annotations

import json
import re
import warnings
from datetime import datetime, timezone
from typing import TYPE_CHECKING, Any, Iterator, List, Optional, Union

from langchain_core.documents import Document

from langchain_community.document_loaders.base import BaseLoader

    import arcgis

_NOT_PROVIDED = "(Not Provided)"

[docs]class ArcGISLoader(BaseLoader): """Load records from an ArcGIS FeatureLayer."""
[docs] def __init__( self, layer: Union[str, arcgis.features.FeatureLayer], gis: Optional[arcgis.gis.GIS] = None, where: str = "1=1", out_fields: Optional[Union[List[str], str]] = None, return_geometry: bool = False, result_record_count: Optional[int] = None, lyr_desc: Optional[str] = None, **kwargs: Any, ): try: import arcgis except ImportError as e: raise ImportError( "arcgis is required to use the ArcGIS Loader. " "Install it with pip or conda." ) from e try: from bs4 import BeautifulSoup self.BEAUTIFULSOUP = BeautifulSoup except ImportError: warnings.warn("BeautifulSoup not found. HTML will not be parsed.") self.BEAUTIFULSOUP = None self.gis = gis or arcgis.gis.GIS() if isinstance(layer, str): self.url = layer self.layer = arcgis.features.FeatureLayer(layer, gis=gis) else: self.url = layer.url self.layer = layer self.layer_properties = self._get_layer_properties(lyr_desc) self.where = where if isinstance(out_fields, str): self.out_fields = out_fields elif out_fields is None: self.out_fields = "*" else: self.out_fields = ",".join(out_fields) self.return_geometry = return_geometry self.result_record_count = result_record_count self.return_all_records = not isinstance(result_record_count, int) query_params = dict( where=self.where, out_fields=self.out_fields, return_geometry=self.return_geometry, return_all_records=self.return_all_records, result_record_count=self.result_record_count, ) query_params.update(kwargs) self.query_params = query_params
def _get_layer_properties(self, lyr_desc: Optional[str] = None) -> dict: """Get the layer properties from the FeatureLayer.""" import arcgis layer_number_pattern = re.compile(r"/\d+$") props = if lyr_desc is None: # retrieve description from the FeatureLayer if not provided try: if self.BEAUTIFULSOUP: lyr_desc = self.BEAUTIFULSOUP(props["description"]).text else: lyr_desc = props["description"] lyr_desc = lyr_desc or _NOT_PROVIDED except KeyError: lyr_desc = _NOT_PROVIDED try: item_id = props["serviceItemId"] item = self.gis.content.get(item_id) or arcgis.features.FeatureLayer( re.sub(layer_number_pattern, "", self.url), ) try: raw_desc = item.description except AttributeError: raw_desc = if self.BEAUTIFULSOUP: item_desc = self.BEAUTIFULSOUP(raw_desc).text else: item_desc = raw_desc item_desc = item_desc or _NOT_PROVIDED except KeyError: item_desc = _NOT_PROVIDED return { "layer_description": lyr_desc, "item_description": item_desc, "layer_properties": props, }
[docs] def lazy_load(self) -> Iterator[Document]: """Lazy load records from FeatureLayer.""" query_response = self.layer.query(**self.query_params) features = (feature.as_dict for feature in query_response) for feature in features: attributes = feature["attributes"] page_content = json.dumps(attributes) metadata = { "accessed": f"{}Z", "name": self.layer_properties["layer_properties"]["name"], "url": self.url, "layer_description": self.layer_properties["layer_description"], "item_description": self.layer_properties["item_description"], "layer_properties": self.layer_properties["layer_properties"], } if self.return_geometry: try: metadata["geometry"] = feature["geometry"] except KeyError: warnings.warn( "Geometry could not be retrieved from the feature layer." ) yield Document(page_content=page_content, metadata=metadata)