codekingpro/portable-devtools
114k
1import json2import urllib.request3from typing import List4 5from langchain_core.documents import Document6from langchain_core.utils import stringify_dict7 8from langchain_community.document_loaders.base import BaseLoader9 10SPREEDLY_ENDPOINTS = {11 "gateways_options": "https://core.spreedly.com/v1/gateways_options.json",12 "gateways": "https://core.spreedly.com/v1/gateways.json",13 "receivers_options": "https://core.spreedly.com/v1/receivers_options.json",14 "receivers": "https://core.spreedly.com/v1/receivers.json",15 "payment_methods": "https://core.spreedly.com/v1/payment_methods.json",16 "certificates": "https://core.spreedly.com/v1/certificates.json",17 "transactions": "https://core.spreedly.com/v1/transactions.json",18 "environments": "https://core.spreedly.com/v1/environments.json",19}20 21 22class SpreedlyLoader(BaseLoader):23 """Load from `Spreedly` API."""24 25 def __init__(self, access_token: str, resource: str) -> None:26 """Initialize with an access token and a resource.27 28 Args:29 access_token: The access token.30 resource: The resource.31 """32 self.access_token = access_token33 self.resource = resource34 self.headers = {35 "Authorization": f"Bearer {self.access_token}",36 "Accept": "application/json",37 }38 39 def _make_request(self, url: str) -> List[Document]:40 request = urllib.request.Request(url, headers=self.headers)41 42 with urllib.request.urlopen(request) as response:43 json_data = json.loads(response.read().decode())44 text = stringify_dict(json_data)45 metadata = {"source": url}46 return [Document(page_content=text, metadata=metadata)]47 48 def _get_resource(self) -> List[Document]:49 endpoint = SPREEDLY_ENDPOINTS.get(self.resource)50 if endpoint is None:51 return []52 return self._make_request(endpoint)53 54 def load(self) -> List[Document]:55 return self._get_resource()56 