Spaces:
Running
Running
File size: 2,638 Bytes
a55ea57 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 | """
File: /structured_data.py
Created Date: Saturday December 6th 2025
Author: Christian Nonis <alch.infoemail@gmail.com>
-----
Last Modified: Sunday December 21st 2025 11:57:01 am
Modified By: Christian Nonis <alch.infoemail@gmail.com>
-----
"""
import asyncio
from typing import Optional
from fastapi import HTTPException
from starlette.responses import JSONResponse
from src.services.data.main import data_adapter
async def get_structured_data_by_id(id: str, brain_id: str = "default"):
"""
Get structured data by ID.
"""
def _get_data():
data = data_adapter.get_structured_data_by_id(id, brain_id)
if not data:
raise HTTPException(status_code=404, detail="Structured data not found")
return data
result = await asyncio.to_thread(_get_data)
return JSONResponse(
content={
"message": "Structured data retrieved successfully",
"data": result.model_dump(mode="json"),
}
)
async def get_structured_data_list(
limit: int = 10,
skip: int = 0,
types: Optional[list[str]] = None,
query_text: Optional[str] = None,
brain_id: str = "default",
):
"""
Get a list of structured data.
"""
def _get_list():
return data_adapter.get_structured_data_list(
brain_id, limit, skip, types, query_text
)
results, total = await asyncio.to_thread(_get_list)
return JSONResponse(
content={
"message": "Structured data list retrieved successfully",
"data": [r.model_dump(mode="json") for r in results],
"count": len(results),
"total": total,
}
)
async def get_structured_data_types(
brain_id: str = "default",
):
"""
Get all unique types from structured data.
"""
def _get_types():
return data_adapter.get_structured_data_types(brain_id)
results = await asyncio.to_thread(_get_types)
return JSONResponse(
content={
"message": "Structured data types retrieved successfully",
"types": results,
"count": len(results),
}
)
async def get_text_chunks(
brain_id: str = "default",
limit: int = 10,
skip: int = 0,
query_text: Optional[str] = None,
metadata_eq: Optional[dict[str, str]] = None,
order: str = "desc",
):
"""
Get text chunks by a query text and optional metadata equality filters.
"""
return await asyncio.to_thread(
data_adapter.get_text_chunks,
brain_id,
limit,
skip,
query_text,
metadata_eq,
order,
)
|