sql_loader

Suggest edits

Bases: BaseLoader[str, list[RowResult], ListArtifact[TextArtifact]]

Source Code in griptape/loaders/sql_loader.py
@define
class SqlLoader(BaseLoader[str, list[BaseSqlDriver.RowResult], ListArtifact[TextArtifact]]):
    sql_driver: BaseSqlDriver = field(kw_only=True)
    format_row: Callable[[dict], str] = field(
        default=lambda value: "\n".join(f"{key}: {val}" for key, val in value.items()), kw_only=True
    )

    def fetch(self, source: str) -> list[BaseSqlDriver.RowResult]:
        return self.sql_driver.execute_query(source) or []

    def parse(self, data: list[BaseSqlDriver.RowResult]) -> ListArtifact[TextArtifact]:
        return ListArtifact([TextArtifact(self.format_row(row.cells)) for row in data])

  • format_row = field(default=lambda value: '\n'.join(f'{key}: {val}' for (key, val) in value.items()), kw_only=True) class-attribute instance-attribute

  • sql_driver = field(kw_only=True) class-attribute instance-attribute

fetch(source)

Source Code in griptape/loaders/sql_loader.py
def fetch(self, source: str) -> list[BaseSqlDriver.RowResult]:
    return self.sql_driver.execute_query(source) or []

parse(data)

Source Code in griptape/loaders/sql_loader.py
def parse(self, data: list[BaseSqlDriver.RowResult]) -> ListArtifact[TextArtifact]:
    return ListArtifact([TextArtifact(self.format_row(row.cells)) for row in data])
← Prev

pdf_loader

↑ Up

loaders

Next →

text_loader

Could this page be better? Report a problem or suggest an addition!