Method export_to_csv

py/core/providers/database/graphs.py:563–680

Creates a CSV file from the PostgreSQL data and returns the path to the temp file.

(
        self,
        parent_id: UUID,
        store_type: StoreType,
        columns: Optional[list[str]] = None,
        filters: Optional[dict] = None,
        include_header: bool = True,
    )

        )

    async def export_to_csv(
        self,
        parent_id: UUID,
        store_type: StoreType,
        columns: Optional[list[str]] = None,
        filters: Optional[dict] = None,
        include_header: bool = True,
    ) -> tuple[str, IO]:
        """Creates a CSV file from the PostgreSQL data and returns the path to
        the temp file."""
        valid_columns = {
            "id",
            "name",
            "category",
            "description",
            "parent_id",
            "chunk_ids",
            "metadata",
            "created_at",
            "updated_at",
        }

        if not columns:
            columns = list(valid_columns)
        elif invalid_cols := set(columns) - valid_columns:
            raise ValueError(f"Invalid columns: {invalid_cols}")

        select_stmt = f"""
            SELECT
                id::text,
                name,
                category,
                description,
                parent_id::text,
                chunk_ids::text,
                metadata::text,
                to_char(created_at, 'YYYY-MM-DD HH24:MI:SS') AS created_at,
                to_char(updated_at, 'YYYY-MM-DD HH24:MI:SS') AS updated_at
            FROM {self._get_table_name(self._get_entity_table_for_store(store_type))}
        """

        conditions = ["parent_id = $1"]
        params: list[Any] = [parent_id]
        param_index = 2

        if filters:
            for field, value in filters.items():
                if field not in valid_columns:
                    continue

                if isinstance(value, dict):
                    for op, val in value.items():
                        if op == "$eq":
                            conditions.append(f"{field} = ${param_index}")
                            params.append(val)
                            param_index += 1
                        elif op == "$gt":
                            conditions.append(f"{field} > ${param_index}")