Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions robosystems_client/clients/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
OperatorQueryRequest,
OperatorOptions,
QueuedOperatorError,
OperatorRunError,
)
from .operation_client import (
OperationClient,
Expand Down Expand Up @@ -144,6 +145,7 @@
"OperatorQueryRequest",
"OperatorOptions",
"QueuedOperatorError",
"OperatorRunError",
# Operation Client
"OperationClient",
"OperationStatus",
Expand Down
27 changes: 17 additions & 10 deletions robosystems_client/clients/auth_integration.py
Original file line number Diff line number Diff line change
Expand Up @@ -82,9 +82,14 @@ def authenticated_client(self) -> AuthenticatedClient:
def execute_cypher_query(
self, graph_id: str, query: str, parameters: Dict[str, Any] = None
):
"""Execute Cypher query using authenticated SDK client"""
"""Execute Cypher query using authenticated SDK client.

A query the API queues (202) is followed over its SSE stream to the
result, as :meth:`QueryClient.execute_query` does.
"""
from ..api.query.execute_cypher import sync_detailed
from ..models.cypher_statement_request import CypherStatementRequest
from .query_client import QueryOptions, QueryResult, _interpret_response

request = CypherStatementRequest(query=query, parameters=parameters or {})

Expand All @@ -95,15 +100,17 @@ def execute_cypher_query(
body=request,
)

if response.parsed:
return {
"data": getattr(response.parsed, "data", []),
"columns": getattr(response.parsed, "columns", []),
"row_count": getattr(response.parsed, "row_count", 0),
"execution_time_ms": getattr(response.parsed, "execution_time_ms", 0),
}
else:
raise Exception(f"Query failed: {response.status_code}")
outcome = _interpret_response(response, graph_id)
if not isinstance(outcome, QueryResult):
outcome = self.query._wait_for_query_completion(
outcome.operation_id, QueryOptions()
)
return {
"data": outcome.data,
"columns": outcome.columns,
"row_count": outcome.row_count,
"execution_time_ms": outcome.execution_time_ms,
}


class CookieAuthClients(RoboSystemsClients):
Expand Down
6 changes: 4 additions & 2 deletions robosystems_client/clients/document_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
from ..models.search_request import SearchRequest
from ..models.search_response import SearchResponse
from ..types import UNSET
from .token_utils import current_token


class DocumentClient:
Expand All @@ -43,11 +44,12 @@ def __init__(self, config: Dict[str, Any]):
self.timeout = config.get("timeout", 60)

def _get_client(self) -> AuthenticatedClient:
if not self.token:
token = current_token(self.config, self.token)
if not token:
raise Exception("No API key provided. Set X-API-Key in headers.")
return retrying_authenticated_client(
base_url=self.base_url,
token=self.token,
token=token,
headers=self.headers,
config=self.config,
)
Expand Down
1 change: 1 addition & 0 deletions robosystems_client/clients/facade.py
Original file line number Diff line number Diff line change
Expand Up @@ -161,6 +161,7 @@ def __init__(self, config: RoboSystemsClientConfig = None):
"max_retries": config.max_retries,
"retry_delay": config.retry_delay,
"timeout": config.timeout,
"token_provider": config.token_provider,
}

# Initialize async clients
Expand Down
49 changes: 32 additions & 17 deletions robosystems_client/clients/file_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,10 +29,17 @@
from ..models.file_upload_request import FileUploadRequest
from ..models.ingest_file_op import IngestFileOp
from ..models.delete_file_op import DeleteFileOp
from ..types import Unset
from .token_utils import current_token

logger = logging.getLogger(__name__)


def _unset_to_none(value: Any) -> Any:
"""A generated-model field as a plain value: ``UNSET`` becomes ``None``."""
return None if isinstance(value, Unset) else value


@dataclass
class FileUploadOptions:
"""Options for file upload operations"""
Expand Down Expand Up @@ -139,12 +146,14 @@ def upload(

from .retry import retrying_authenticated_client

if not self.token:
token = current_token(self.config, self.token)

if not token:
raise Exception("No API key provided. Set X-API-Key in headers.")

client = retrying_authenticated_client(
base_url=self.base_url,
token=self.token,
token=token,
headers=self.headers,
config=self.config,
)
Expand Down Expand Up @@ -295,12 +304,14 @@ def list(
try:
from .retry import retrying_authenticated_client

if not self.token:
token = current_token(self.config, self.token)

if not token:
raise Exception("No API key provided. Set X-API-Key in headers.")

client = retrying_authenticated_client(
base_url=self.base_url,
token=self.token,
token=token,
headers=self.headers,
config=self.config,
)
Expand Down Expand Up @@ -330,11 +341,11 @@ def list(
file_name=f.file_name,
file_format=f.file_format,
size_bytes=f.size_bytes or 0,
row_count=f.row_count,
row_count=_unset_to_none(f.row_count),
upload_status=f.upload_status,
table_name=getattr(f, "table_name", ""),
created_at=f.created_at,
uploaded_at=f.uploaded_at,
table_name=_unset_to_none(getattr(f, "table_name", None)) or "",
created_at=_unset_to_none(f.created_at),
uploaded_at=_unset_to_none(f.uploaded_at),
)
for f in files
]
Expand All @@ -357,12 +368,14 @@ def get(self, graph_id: str, file_id: str) -> Optional[FileInfo]:
try:
from .retry import retrying_authenticated_client

if not self.token:
token = current_token(self.config, self.token)

if not token:
raise Exception("No API key provided. Set X-API-Key in headers.")

client = retrying_authenticated_client(
base_url=self.base_url,
token=self.token,
token=token,
headers=self.headers,
config=self.config,
)
Expand All @@ -386,12 +399,12 @@ def get(self, graph_id: str, file_id: str) -> Optional[FileInfo]:
file_name=file_data.file_name,
file_format=file_data.file_format,
size_bytes=file_data.size_bytes or 0,
row_count=file_data.row_count,
row_count=_unset_to_none(file_data.row_count),
upload_status=file_data.upload_status,
table_name=file_data.table_name or "",
created_at=file_data.created_at,
uploaded_at=file_data.uploaded_at,
layers=getattr(file_data, "layers", None),
table_name=_unset_to_none(file_data.table_name) or "",
created_at=_unset_to_none(file_data.created_at),
uploaded_at=_unset_to_none(file_data.uploaded_at),
layers=_unset_to_none(getattr(file_data, "layers", None)),
)

except Exception as e:
Expand All @@ -413,12 +426,14 @@ def delete(self, graph_id: str, file_id: str, cascade: bool = False) -> bool:
try:
from .retry import retrying_authenticated_client

if not self.token:
token = current_token(self.config, self.token)

if not token:
raise Exception("No API key provided. Set X-API-Key in headers.")

client = retrying_authenticated_client(
base_url=self.base_url,
token=self.token,
token=token,
headers=self.headers,
config=self.config,
)
Expand Down
55 changes: 33 additions & 22 deletions robosystems_client/clients/graph_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,11 @@
automatic operation monitoring. Supports both SSE (Server-Sent Events)
for real-time updates and polling fallback.

Graph lifecycle operations (create-subgraph, delete-subgraph, create-backup,
change-tier, materialize) all go through the operations surface at
The lifecycle operations this client wraps (materialize, change-tier) go
through the operations surface at
``POST /v1/graphs/{graph_id}/operations/{op_name}`` and return an
``OperationEnvelope``.
``OperationEnvelope``. Subgraph and backup operations live on the same
surface but have no method here — call the generated ``api`` functions.

Backups are a download capability: there is no customer-facing restore.
Graphs with an upstream rebuild from it — entity graphs re-materialize,
Expand All @@ -23,7 +24,9 @@

import httpx

from ..types import Unset
from .operation_client import OperationClient, OperationProgress, MonitorOptions
from .token_utils import apply_auth_header, current_token

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -96,8 +99,8 @@ class MaterializationResult:
class GraphClient:
"""Client for graph management and lifecycle operations.

Covers graph creation, info retrieval, and all graph lifecycle
operations (materialize, subgraphs, backups, tier changes).
Covers graph creation and deletion, info retrieval, materialization
and tier changes.
"""

def __init__(self, config: Dict[str, Any]):
Expand All @@ -118,12 +121,14 @@ def _get_authenticated_client(self):
"""Build an AuthenticatedClient for API calls."""
from .retry import retrying_authenticated_client

if not self.token:
token = current_token(self.config, self.token)

if not token:
raise ValueError("No API key provided. Set X-API-Key in headers.")

return retrying_authenticated_client(
base_url=self.base_url,
token=self.token,
token=token,
headers=self.headers,
config=self.config,
)
Expand Down Expand Up @@ -250,18 +255,18 @@ def get_graph_info(self, graph_id: str) -> GraphInfo:
client = self._get_authenticated_client()
response = get_graphs(client=client)

if not response.parsed:
if response.status_code != 200 or not response.parsed:
raise RuntimeError(f"Failed to get graphs: {response.status_code}")

# Every generated model carries `additional_properties` (the fields the
# schema does not name), so the declared `graphs` field is read first.
data = response.parsed
graphs = None

if isinstance(data, dict):
graphs = data.get("graphs", [])
elif hasattr(data, "additional_properties"):
graphs = data.additional_properties.get("graphs", [])
elif hasattr(data, "graphs"):
graphs = data.graphs
elif hasattr(data, "additional_properties"):
graphs = data.additional_properties.get("graphs", [])
else:
raise RuntimeError("Unexpected response format from get_graphs")

Expand Down Expand Up @@ -290,16 +295,19 @@ def get_graph_info(self, graph_id: str) -> GraphInfo:
status=graph_data.get("status"),
)
else:

def field(name: str) -> Any:
value = getattr(graph_data, name, None)
return None if isinstance(value, Unset) else value

return GraphInfo(
graph_id=getattr(graph_data, "graph_id", None)
or getattr(graph_data, "id", graph_id),
graph_name=getattr(graph_data, "graph_name", None)
or getattr(graph_data, "name", ""),
description=getattr(graph_data, "description", None),
schema_extensions=getattr(graph_data, "schema_extensions", None),
tags=getattr(graph_data, "tags", None),
created_at=getattr(graph_data, "created_at", None),
status=getattr(graph_data, "status", None),
graph_id=field("graph_id") or field("id") or graph_id,
graph_name=field("graph_name") or field("name") or "",
description=field("description"),
schema_extensions=field("schema_extensions"),
tags=field("tags"),
created_at=field("created_at"),
status=field("status"),
)

# ---------------------------------------------------------------------------
Expand Down Expand Up @@ -535,7 +543,10 @@ def _wait_with_sse(
) -> str:
"""Wait for operation completion using SSE stream."""
stream_url = f"{self.base_url}/v1/operations/{operation_id}/stream"
headers = {"X-API-Key": self.token, "Accept": "text/event-stream"}
headers = {"Accept": "text/event-stream"}
token = current_token(self.config, self.token)
if token:
apply_auth_header(headers, token)

with httpx.Client(timeout=httpx.Timeout(timeout + 5.0)) as http_client:
with http_client.stream("GET", stream_url, headers=headers) as response:
Expand Down
Loading
Loading