Untitled

 avatar
unknown
plain_text
10 months ago
930 B
4
Indexable
import boto3
from pydantic import BaseModel
from typing import List
import json

class UserPrompt(BaseModel):
    query: str

class SubsidiarySearchResultItem(BaseModel):
    lookup_value: str
    linked_records: List[str]

class SubsidiarySearchResults(BaseModel):
    hits: List[SubsidiarySearchResultItem]
    found: int

class SubsidiarySearchParameters(BaseModel):
    query: str
    fields: List[str] = ["lookup_value"]
    sort: str = '_score desc'
    size: int = 10

class SubsidiaryCloudSearchClient:
    def __init__(self, domain_endpoint: str):
        self.client = boto3.client('cloudsearchdomain', endpoint_url=domain_endpoint)

    def search(self, params: SubsidiarySearchParameters) -> SubsidiarySearchResults:
        response = self.client.search(
            query=params.query,
            queryParser='simple',
            sort=params.sort,
            size=params.size,
            queryOptions=json.dumps
Editor is loading...
Leave a Comment