repositories
loading repo index
repositories
loading repo index
repository
loading code, commits, and activity
public Clawd ADK gateway launch mirror
stars
latest
clone command
git clone gitlawb://did:key:z6Mkq5mY...iFZ5/my-project-publ...git clone gitlawb://did:key:z6Mkq5mY.../my-project-publ...2fa351d6docs: add automaton and perps launch sources16d ago| #1 | from collections.abc import Callable |
| #2 | from typing import Any, Dict, List, Optional |
| #3 | |
| #4 | from pydantic import BaseModel, Field, model_validator |
| #5 | |
| #6 | |
| #7 | class ElasticsearchConfig(BaseModel): |
| #8 | collection_name: str = Field("mem0", description="Name of the index") |
| #9 | host: str = Field("localhost", description="Elasticsearch host") |
| #10 | port: int = Field(9200, description="Elasticsearch port") |
| #11 | user: Optional[str] = Field(None, description="Username for authentication") |
| #12 | password: Optional[str] = Field(None, description="Password for authentication") |
| #13 | cloud_id: Optional[str] = Field(None, description="Cloud ID for Elastic Cloud") |
| #14 | api_key: Optional[str] = Field(None, description="API key for authentication") |
| #15 | embedding_model_dims: int = Field(1536, description="Dimension of the embedding vector") |
| #16 | verify_certs: bool = Field(True, description="Verify SSL certificates") |
| #17 | use_ssl: bool = Field(True, description="Use SSL for connection") |
| #18 | auto_create_index: bool = Field(True, description="Automatically create index during initialization") |
| #19 | custom_search_query: Optional[Callable[[List[float], int, Optional[Dict]], Dict]] = Field( |
| #20 | None, description="Custom search query function. Parameters: (query, limit, filters) -> Dict" |
| #21 | ) |
| #22 | headers: Optional[Dict[str, str]] = Field(None, description="Custom headers to include in requests") |
| #23 | |
| #24 | @model_validator(mode="before") |
| #25 | @classmethod |
| #26 | def validate_auth(cls, values: Dict[str, Any]) -> Dict[str, Any]: |
| #27 | # Check if either cloud_id or host/port is provided |
| #28 | if not values.get("cloud_id") and not values.get("host"): |
| #29 | raise ValueError("Either cloud_id or host must be provided") |
| #30 | |
| #31 | # Check if authentication is provided |
| #32 | if not any([values.get("api_key"), (values.get("user") and values.get("password"))]): |
| #33 | raise ValueError("Either api_key or user/password must be provided") |
| #34 | |
| #35 | return values |
| #36 | |
| #37 | @model_validator(mode="before") |
| #38 | @classmethod |
| #39 | def validate_headers(cls, values: Dict[str, Any]) -> Dict[str, Any]: |
| #40 | """Validate headers format and content""" |
| #41 | headers = values.get("headers") |
| #42 | if headers is not None: |
| #43 | # Check if headers is a dictionary |
| #44 | if not isinstance(headers, dict): |
| #45 | raise ValueError("headers must be a dictionary") |
| #46 | |
| #47 | # Check if all keys and values are strings |
| #48 | for key, value in headers.items(): |
| #49 | if not isinstance(key, str) or not isinstance(value, str): |
| #50 | raise ValueError("All header keys and values must be strings") |
| #51 | |
| #52 | return values |
| #53 | |
| #54 | @model_validator(mode="before") |
| #55 | @classmethod |
| #56 | def validate_extra_fields(cls, values: Dict[str, Any]) -> Dict[str, Any]: |
| #57 | allowed_fields = set(cls.model_fields.keys()) |
| #58 | input_fields = set(values.keys()) |
| #59 | extra_fields = input_fields - allowed_fields |
| #60 | if extra_fields: |
| #61 | raise ValueError( |
| #62 | f"Extra fields not allowed: {', '.join(extra_fields)}. " |
| #63 | f"Please input only the following fields: {', '.join(allowed_fields)}" |
| #64 | ) |
| #65 | return values |
| #66 |