Files
langchain/libs/community/langchain_community/utilities/mojeek_search.py
Harrison Chase 8516a03a02 langchain-community[major]: Upgrade community to pydantic 2 (#26011)
This PR upgrades langchain-community to pydantic 2.


* Most of this PR was auto-generated using code mods with gritql
(https://github.com/eyurtsev/migrate-pydantic/tree/main)
* Subsequently, some code was fixed manually due to accommodate
differences between pydantic 1 and 2

Breaking Changes:

- Use TEXTEMBED_API_KEY and TEXTEMBEB_API_URL for env variables for text
embed integrations:
cbea780492

Other changes:

- Added pydantic_settings as a required dependency for community. This
may be removed if we have enough time to convert the dependency into an
optional one.

---------

Co-authored-by: Eugene Yurtsev <eyurtsev@gmail.com>
Co-authored-by: Bagatur <baskaryan@gmail.com>
2024-09-05 14:07:10 -04:00

45 lines
1.3 KiB
Python

import json
from typing import List
import requests
from pydantic import BaseModel, Field
class MojeekSearchAPIWrapper(BaseModel):
api_key: str
search_kwargs: dict = Field(default_factory=dict)
api_url: str = "https://api.mojeek.com/search"
def run(self, query: str) -> str:
search_results = self._search(query)
results = []
for result in search_results:
title = result.get("title", "")
url = result.get("url", "")
desc = result.get("desc", "")
results.append({"title": title, "url": url, "desc": desc})
return json.dumps(results)
def _search(self, query: str) -> List[dict]:
headers = {
"Accept": "application/json",
}
req = requests.PreparedRequest()
request = {
**self.search_kwargs,
**{"q": query, "fmt": "json", "api_key": self.api_key},
}
req.prepare_url(self.api_url, request)
if req.url is None:
raise ValueError("prepared url is None, this should not happen")
response = requests.get(req.url, headers=headers)
if not response.ok:
raise Exception(f"HTTP error {response.status_code}")
return response.json().get("response", {}).get("results", [])