tools/database/types.py (53 lines of code) (raw):

# Database Tools Models from pydantic import BaseModel, Field from typing import Dict, List, Optional, Union, Literal class DataSourceItem(BaseModel): """ Represents a data source with its type information. Attributes: name: The name of the data source. description: The description of the data source. type: The type of the data source. """ name: str description: str type: str class DataSourcesList(BaseModel): """ Represents a list of available data sources. Attributes: datasources: A list of DataSourceItem instances, each containing information about a data source. """ datasources: List[DataSourceItem] class TableItem(BaseModel): """ Represents information about a specific database table. Attributes: table: The name of the table. description: A brief description of the table. datasource: The name of the data source where the table resides. """ table: str description: str datasource: str class TablesList(BaseModel): """ Represents a list of tables along with optional error information. Attributes: tables: A list of TableItem instances, each representing a table. error: An optional error message, if any issues were encountered. """ tables: List[TableItem] error: Optional[str] = None class SchemaInfo(BaseModel): """ Represents the schema details of a database table. Attributes: datasource: The name of the data source where the table resides. table: The name of the table. description: An optional description of the table. columns: A dictionary mapping column names to their respective descriptions. """ datasource: str table: str description: Optional[str] = None columns: Optional[Dict[str, str]] = None # Map column names to descriptions class ValidateSQLQueryResult(BaseModel): """ Represents the result of a SQL query validation. Attributes: is_valid: Indicates whether the SQL query is valid. error: An optional error message if the query is invalid. """ is_valid: bool error: Optional[str] = None class ExecuteQueryResult(BaseModel): """ Represents the result of executing a SQL query. Attributes: results: A list of dictionaries representing the query results. Each dictionary maps column names to their respective values. error: An optional error message if the query execution failed. """ results: Optional[List[Dict[str, Union[str, int, float, None]]]] = None error: Optional[str] = None # Database AI Search Index Retrieval Models from pydantic import BaseModel, Field from typing import List, Optional class QueryItem(BaseModel): """ Represents a single query retrieval result. Attributes: question: The question from the search result. query: The optimized query string. reasoning: Explanation or reasoning behind the query construction. """ question: str = Field(..., description="The question from the search result") query: str = Field(..., description="The optimized query string") reasoning: str = Field(..., description="The reasoning behind the query construction") class QueriesRetrievalResult(BaseModel): """ Represents the overall result for queries retrieval. Attributes: queries (List[QueryItem]): A list of query retrieval results. error (Optional[str]): An error message, if any. Defaults to None. """ queries: List[QueryItem] = Field(..., description="A list of query retrieval results") error: Optional[str] = Field(None, description="Error message if query fails") # For tables_retrieval class TableRetrievalItem(BaseModel): """ Represents a single table entry with its name, description, and datasource. """ table: str = Field(..., description="The name of the table") description: str = Field(..., description="A brief description of the table") datasource: Optional[str] = Field(None, description="The datasource used for retrieval") class TablesRetrievalResult(BaseModel): """ Represents the result for tables retrieval. Attributes: tables: A list of TableRetrievalItem objects. error (Optional[str]): An error message, if any. Defaults to None. """ tables: List[TableRetrievalItem] = Field(..., description="List of tables with details") error: Optional[str] = Field(None, description="Error message if query fails") class MeasureItem(BaseModel): """ Represents information about a specific measure. Attributes: name: The name of the measure. description: A brief description of the measure. datasource: The datasource where the measure resides. type: The type of the measure ("external" or "local"). source_table: The source table associated with the measure. data_type: The data type of the measure. source_model: The source model for the measure. """ name: str = Field(..., description="The name of the measure") description: str = Field(..., description="A brief description of the measure") datasource: str = Field(..., description="The datasource for the measure") type: Literal["external", "local"] = Field(..., description="The type of the measure (external or local)") source_table: Optional[str] = Field(None, description="The source table for the measure") data_type: Optional[str] = Field(None, description="The data type of the measure") source_model: Optional[str] = Field(None, description="The source model for the external measure") class MeasuresList(BaseModel): """ Represents a list of measures along with optional error information. Attributes: measures: A list of MeasureItem instances. error: An optional error message if issues occur. """ measures: List[MeasureItem] = Field(..., description="List of measures with details") error: Optional[str] = Field(None, description="Error message if query fails")