diff --git a/README.md b/README.md index 2a13233..135a6da 100644 --- a/README.md +++ b/README.md @@ -3,16 +3,16 @@ [![PyPI version](https://badge.fury.io/py/robosystems-client.svg)](https://pypi.org/project/robosystems-client/) [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT) -Official Python client for the RoboSystems financial intelligence platform — accounting, financial reporting, and investment management over a knowledge graph API. Type-safe and async-ready, generated from the OpenAPI spec, with GraphQL reads and idempotent operation writes. +Official Python client for the RoboSystems financial intelligence platform — accounting, financial reporting, and investment management over a knowledge graph API. Type-safe, generated from the OpenAPI spec, with GraphQL reads and idempotent operation writes. ## Features - **Type-safe API client** with full type hints and typed models -- **Async/await support** for high-performance applications +- **Async client** for queries and operations - **Parquet file uploads** for table ingestion -- **Streaming support** for memory-efficient processing of large result sets +- **Row iteration** over query results with `stream_query`, which collects the full result before yielding rows - **AI Operator** integration for natural language financial analysis -- **Comprehensive error handling** with typed exceptions +- **Typed errors** for GraphQL reads and queued work ## Installation diff --git a/robosystems_client/clients/README.md b/robosystems_client/clients/README.md index 2d3d97e..97fceff 100644 --- a/robosystems_client/clients/README.md +++ b/robosystems_client/clients/README.md @@ -1,634 +1,263 @@ -# RoboSystems Python Client Extensions +# RoboSystems Python Client: Facades -**Production-Ready Extensions** for the RoboSystems financial intelligence platform +`robosystems_client.clients` is the hand-written layer over the generated `api/` and `models/` packages and the GraphQL client. It sets auth headers, follows queued operations to a result, and replays rate-limited requests. -[![Python 3.7+](https://img.shields.io/badge/python-3.7+-blue.svg)](https://www.python.org/downloads/) -[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT) +Install with `pip install robosystems-client`. It requires Python 3.12 or newer. `httpx` is a core dependency. `pandas` is optional and only the DataFrame helpers need it. Tests live in `tests/` at the repository root; run them with `just test`. -## Overview - -The RoboSystems Python Client Extensions provide enhanced functionality for the auto-generated Python Client, adding: - -- **Server-Sent Events (SSE)** streaming with automatic reconnection -- **Smart Query Execution** with automatic strategy selection -- **Operation Monitoring** for long-running operations -- **Connection Pooling** and intelligent resource management -- **Result Processing** and format conversion utilities -- **Caching** with TTL and LRU eviction -- **Full Async/Await Support** throughout - -## Quick Start - -### Installation - -```bash -# Install the extensions (part of the main client) -pip install robosystems-client - -# Optional dependencies for enhanced features -pip install httpx pandas # For SSE and DataFrame support -``` - -### Basic Usage +## Quick start ```python -from robosystems_client.clients import create_production_clients +from robosystems_client.clients import AuthenticatedClients -# Initialize with API key -extensions = create_production_clients("your-api-key-here") +clients = AuthenticatedClients("rfs-your-api-key") -# Execute a simple query -result = extensions.execute_query( - graph_id="your_graph_id", query="MATCH (c:Company) RETURN c.name, c.revenue LIMIT 10" +result = clients.execute_query( + "sec", + "MATCH (e:Entity) WHERE e.ticker = $ticker RETURN e.name, e.cik LIMIT 1", + {"ticker": "NVDA"}, ) +for row in result.data: + print(row["e.name"], row["e.cik"]) -print(f"Found {result.row_count} companies") -for company in result.data: - print(f"- {company['name']}: ${company['revenue']:,}") - -# Clean up -extensions.close() +clients.close() ``` -### Streaming Large Results - -```python -# Stream large datasets efficiently -query = "MATCH (t:Transaction) RETURN t.id, t.amount, t.timestamp" - -for transaction in extensions.stream_query("your_graph_id", query, chunk_size=1000): - # Process each transaction as it arrives - process_transaction(transaction) -``` - -### Async Support - -```python -import asyncio -from robosystems_client.clients import AsyncRoboSystemsClients - - -async def main(): - extensions = AsyncRoboSystemsClients(config) +Reading the shared `sec` graph needs a key with access to it. Later examples continue from a `clients` built this way and use placeholders such as `graph_id` and `operation_id`. - # Async query execution - result = await extensions.execute_query("graph_id", "MATCH (n) RETURN COUNT(n)") - print(f"Total nodes: {result.data[0]['COUNT(n)']}") +## What a client holds - # Async streaming - async for row in extensions.stream_query("graph_id", "MATCH (n) RETURN n"): - await process_row_async(row) +| Attribute | Class | Covers | +| --------- | ----- | ------ | +| `query` | `QueryClient` | Cypher: `query`, `execute_query`, `stream_query`, `query_batch` | +| `operator` | `OperatorClient` | AI Operator runs: `query`, `execute_query`, `execute_operator` | +| `operations` | `OperationClient` | `monitor_operation`, `get_operation_status`, `cancel_operation` | +| `graphs` | `GraphClient` | `create_graph_and_wait`, `get_graph_info`, `materialize`, `change_tier`, `delete_graph` | +| `files` | `FileClient` | Parquet files: `upload`, `list`, `get`, `delete` | +| `tables` | `TableClient` | Staging tables: `list`, `query` (SQL) | +| `documents` | `DocumentClient` | Markdown documents: `upload`, `upload_file`, `upload_directory`, `update`, `get`, `get_section`, `search`, `list`, `delete` | +| `ledger` | `LedgerClient` | RoboLedger reads and writes (`reports` is the same object) | +| `investor` | `InvestorClient` | RoboInvestor reads and writes | +| `library` | `LibraryClient` | Taxonomy library reads | - await extensions.close() +The top level also has `execute_query`, `stream_query`, `monitor_operation`, `get_operation_status`, `cancel_operation`, `create_sse_client` and `close`. - -asyncio.run(main()) -``` +The package root exports `RoboSystemsClients`, `RoboSystemsClientConfig`, `LedgerClient`, `InvestorClient`, `LibraryClient`, `GraphQLError` and `get_clients`. The other classes and helpers here import from `robosystems_client.clients`, except `DocumentClient` (`robosystems_client.clients.document_client`) and `MonitorOptions` (`robosystems_client.clients.operation_client`). ## Authentication -### API Key Authentication (Recommended) +| Constructor | Credential | Default base URL | +| ----------- | ---------- | ---------------- | +| `AuthenticatedClients(api_key, config=None, base_url=None)` | API key | `https://api.robosystems.ai` | +| `TokenClients(token, config=None, base_url=None)` | JWT | `https://api.robosystems.ai` | +| `RoboSystemsClients(config=None)` | `config.headers` or `config.token_provider` | `http://localhost:8000` | -```python -from robosystems_client.clients import AuthenticatedClients +`create_production_clients(api_key)` and `create_development_clients(api_key)` build `AuthenticatedClients` for `https://api.robosystems.ai` and `http://localhost:8000`. A credential is routed by its shape: a value starting with `rfs` goes in `X-API-Key`, anything else in `Authorization: Bearer`. Only one header is sent. -extensions = AuthenticatedClients( - api_key="your-api-key-here", base_url="https://api.robosystems.ai" -) -``` - -### Cookie Authentication +For a credential that rotates, build `RoboSystemsClients(RoboSystemsClientConfig(base_url=url, token_provider=load_current_jwt))`. The zero-argument callable runs on every request and stream connect and wins over a static credential, so keep it cheap. `create_sse_client()` and `authenticated_client` do not consult it; they hold the credential set at construction. -```python -from robosystems_client.clients import CookieAuthClients +`CookieAuthClients` and `create_clients("cookie", ...)` are not a working way to authenticate. The cookies never reach the child clients, whose calls raise "No API key provided", and the API authenticates data calls by header. `get_clients()` takes no arguments and returns a shared `RoboSystemsClients` with the defaults, localhost and no credential, so build your own client to reach the API. -extensions = CookieAuthClients( - cookies={"auth-token": "your-cookie-token"}, base_url="https://api.robosystems.ai" -) -``` +The package reads no environment variables on its own. `extract_token_from_environment(env_var="ROBOSYSTEMS_TOKEN")` reads one only when you call it, and returns the value only if it is JWT-shaped. -### JWT Token Authentication +## Configuration -```python -from robosystems_client.clients import TokenClients +| `RoboSystemsClientConfig` field | Default | Effect | +| ------------------------------- | ------- | ------ | +| `base_url` | `None` | API root; see the table above for what `None` becomes | +| `headers` | `None` | Sent on API calls; may carry a static `X-API-Key` or `Authorization: Bearer` | +| `max_retries` | `5` | Replays of a `429` response; reconnect attempts for `create_sse_client()` | +| `retry_delay` | `1000` | Base backoff in milliseconds for both of the above | +| `timeout` | `30` | Seconds, for ledger, investor and library GraphQL reads and `create_sse_client()`. Other REST calls set no client timeout | +| `s3_endpoint_url` | `None` | Rewrites the host of file upload URLs, for example for LocalStack | +| `token_provider` | `None` | Callable returning the current credential | -extensions = TokenClients(token="your-jwt-token", base_url="https://api.robosystems.ai") -``` +## Cypher queries -### Rotating Credentials (`token_provider`) +`clients.execute_query(graph_id, query, parameters=None)` and `clients.query.query(...)` return a `QueryResult` with `data`, `columns`, `row_count`, `execution_time_ms`, `graph_id` and `timestamp`. Each row is a dict keyed by column, so `RETURN e.name` gives `row["e.name"]` and `RETURN e.name AS name` gives `row["name"]`. A failed query raises `Exception` with the status and detail in its message. -Short-lived JWTs rotate, and the backend revokes the previous token on every -session refresh — a credential captured when the facade was built stops -working the moment the session rotates. Pass a zero-arg callable instead and -every client resolves the credential fresh: the GraphQL facades (`ledger` / -`investor` / `library`) on each request, and the SSE-backed clients -(`operator` / `operations` / `query`) on each REST call _and_ each stream -connect. It wins over any static `token` or auth header in `headers`, and is -routed by shape (`rfs…` keys as `X-API-Key`, anything else as a Bearer JWT). +**Queued queries.** When the API queues a query, those calls block on the operation's SSE stream until it ends. To get control back, call `QueryClient.execute_query` with `max_wait=0`, the only path that raises `QueuedQueryError`. No other `max_wait` value has an effect. ```python -from robosystems_client.clients import RoboSystemsClients, RoboSystemsClientConfig - -extensions = RoboSystemsClients( - RoboSystemsClientConfig( - base_url="https://api.robosystems.ai", - token_provider=lambda: load_current_jwt(), # or `lambda: manager.token` - ) -) -``` - -`OperatorClient` also follows a queued run over `/v1/operations/{id}/status` -whenever its stream gives no verdict — it could not open, its reconnects ran -out, or it ended before a terminal event — so a run that is already executing -is never lost. `OperatorOptions.poll_interval` (seconds) tunes the interval. - -### Environment-Specific Configurations +from robosystems_client.clients import QueryOptions, QueryRequest, QueuedQueryError -```python -from robosystems_client.clients import create_clients - -# Production -prod_ext = create_clients( - "api_key", - api_key=os.getenv("ROBOSYSTEMS_API_KEY"), - base_url="https://api.robosystems.ai", -) - -# Development -dev_ext = create_clients( - "api_key", api_key="dev-key-123", base_url="http://localhost:8000" -) +request = QueryRequest(query="MATCH (n) RETURN count(n) AS total") +try: + result = clients.query.execute_query(graph_id, request, QueryOptions(max_wait=0)) +except QueuedQueryError as queued: + print(queued.queue_info.operation_id, queued.queue_info.queue_position) ``` -## Advanced Features +**`stream_query`** returns an iterator of rows, but it is not incremental. The facade collects the whole result, from the NDJSON response or from a queued query's SSE stream, and then yields it row by row. Memory use is the size of the full result, and `chunk_size` does not change that. Call it as `clients.stream_query(graph_id, query, parameters=None, chunk_size=1000)`. -### Query Builder +**`clients.query.query_batch(graph_id, queries, parameters_list=None)`** runs queries one after another; its `parallel` argument has no effect. A failed query appears in the list as `{"error": ..., "query": ...}`. -Build complex Cypher queries programmatically: +**`QueryBuilder`** appends clauses in the order you call it, one per line. Each `.where()` adds its own `WHERE` line, so two calls produce invalid Cypher. Use one `.where()` with `AND`. ```python from robosystems_client.clients import QueryBuilder -builder = QueryBuilder() -query, params = ( - builder.match("(c:Company)-[:HAS_TRANSACTION]->(t:Transaction)") - .match("(c)-[:LOCATED_IN]->(l:Location)") - .where("c.revenue > $min_revenue") - .where("l.country = $country") - .return_("c.name", "c.revenue", "COUNT(t) as tx_count", "l.city") - .order_by("tx_count DESC", "c.revenue DESC") - .limit(50) - .with_param("min_revenue", 1000000) - .with_param("country", "USA") - .build() -) - -print("Generated Query:") -print(query) -print("Parameters:", params) -``` - -### Query Validation & Cost Estimation - -```python -from robosystems_client.clients import validate_cypher_query, estimate_query_cost - -# Validate syntax -validation = validate_cypher_query(query) -if not validation["valid"]: - print("Query issues:", validation["issues"]) - -# Estimate complexity -cost = estimate_query_cost(query, params) -print(f"Complexity: {cost['complexity_category']} (score: {cost['complexity_score']})") - -# Get optimization recommendations -for rec in cost["recommendations"]: - print(f"Tip: {rec}") +builder = QueryBuilder().match("(e:Entity)") +builder.where("e.sic = $sic AND e.ticker IS NOT NULL").with_param("sic", "3674") +builder.return_("e.name", "e.ticker").order_by("e.name").limit(10) +query, params = builder.build() +result = clients.execute_query("sec", query, params) ``` -### Result Processing - -Convert results to different formats: - -```python -from robosystems_client.clients import ResultProcessor - -# Convert to JSON -json_output = ResultProcessor.to_json(result, pretty=True) - -# Convert to CSV -csv_output = ResultProcessor.to_csv(result) - -# Convert to pandas DataFrame (if pandas installed) -try: - df = ResultProcessor.to_dataframe(result) - print(df.head()) -except ImportError: - print("pandas not installed") -``` +**Standalone helpers.** No facade uses these automatically; call them yourself. `CacheManager` is an in-memory TTL cache, and `ProgressTracker` and `DataBatcher` are small bookkeeping classes. `ResultProcessor` has `to_json`, `to_csv` and `to_dataframe`. `validate_cypher_query` checks bracket balance and a few patterns and is not a parser. `estimate_query_cost` counts keywords and does not contact the server. `query_result_to_dataframe` and `DataFrameQueryClient` need pandas. -### Caching +## Long-running operations -Cache expensive queries automatically: +`monitor_operation` follows `GET /v1/operations/{id}/stream`, replayed from the start, and returns an `OperationResult`. A failed or cancelled operation comes back as a `status`, not an exception. Without a timeout, it raises only if the stream ends with no terminal event. The optional callback receives an `OperationProgress`, with `message` and `percentage`. ```python -from robosystems_client.clients import CacheManager - -cache = CacheManager(max_size=100, ttl_seconds=300) # 5 minute TTL - -# Cache is used automatically by extensions, or manually: -cached_result = cache.get("graph_id", query, parameters) -if not cached_result: - result = extensions.execute_query("graph_id", query, parameters) - cache.set("graph_id", query, result, parameters) -else: - print("Using cached result!") - -# View cache statistics -print(cache.stats()) +result = clients.monitor_operation(operation_id, on_progress) +print(result.status.value, result.result, result.error) ``` -### Progress Tracking +`status` is an `OperationStatus`. `FAILED` can also mean the stream connection failed while the operation was still running, with the transport error in `error`. `get_operation_status(operation_id)` returns the server's view as a dict, and `cancel_operation(operation_id)` returns a bool. To set a limit, call `clients.operations.monitor_operation(operation_id, MonitorOptions(timeout=600))`, which raises `TimeoutError`. -Monitor long-running operations: +For custom event handling, use `create_sse_client()`, which copies the configured headers. An `SSEConfig` you build yourself needs credentials in `headers`, because the stream endpoint returns 401 without them. `connect` reports problems through the `error` event rather than raising. It retries with backoff only when opening the connection raises; a non-200 response or a drop mid-stream emits `error` and returns. ```python -from robosystems_client.clients import ProgressTracker - - -def progress_handler(progress): - print(f"Step {progress.current_step}/{progress.total_steps}: {progress.message}") - if progress.percentage: - print(f"Progress: {progress.percentage}%") - - -# Monitor operation with progress callback -result = extensions.monitor_operation("operation_id", progress_handler) -print(f"Operation completed: {result.status}") +from robosystems_client.clients import EventType + +sse = clients.create_sse_client() +sse.on(EventType.OPERATION_PROGRESS.value, lambda data: print(data.get("message"))) +sse.on(EventType.OPERATION_COMPLETED.value, lambda data: print(data.get("result"))) +sse.on("error", lambda error: print("stream error:", error)) +sse.connect(operation_id) # blocks until the stream ends +sse.close() ``` -### SSE Streaming +## AI Operator -Direct SSE connection for real-time events: +`clients.operator.query(graph_id, message)` picks an operator and returns an `OperatorResult` with `content`, `operator_used`, `mode_used`, `operation_id` and `error_details`. `error_details` is set when the API reports a failed run inside a successful response. Use `execute_query` for options, or `execute_operator(graph_id, "analyst", request)` to name the operator. ```python -from robosystems_client.clients import SSEClient, SSEConfig, EventType - -config = SSEConfig(base_url="https://api.robosystems.ai") -client = SSEClient(config) - - -# Set up event handlers -def handle_progress(data): - print(f"Progress: {data['message']}") - - -def handle_data_chunk(data): - print(f"Received {len(data.get('rows', []))} rows") - - -client.on(EventType.OPERATION_PROGRESS.value, handle_progress) -client.on(EventType.DATA_CHUNK.value, handle_data_chunk) - -# Connect and monitor -client.connect("operation_id") - -# Clean up -client.close() -``` - -## Examples - -### Financial Data Analysis - -```python -from robosystems_client.clients import create_production_clients, QueryBuilder - -extensions = create_production_clients(api_key) - -# Find top performing companies by revenue growth -builder = QueryBuilder() -query, params = ( - builder.match("(c:Company)-[:HAS_FINANCIAL_REPORT]->(r:FinancialReport)") - .where("r.year >= $start_year") - .where("c.revenue > $min_revenue") - .return_("c.name", "c.revenue", "r.year", "r.revenue_growth_rate") - .order_by("r.revenue_growth_rate DESC") - .limit(20) - .with_param("start_year", 2020) - .with_param("min_revenue", 10000000) - .build() -) - -result = extensions.execute_query("financial_graph", query, params) - -print("Top 20 Companies by Revenue Growth:") -for company in result.data: - print(f"- {company['name']}: {company['revenue_growth_rate']:.1f}% growth") -``` - -### Batch Processing with Streaming - -```python -from robosystems_client.clients import DataBatcher, format_duration -import time - -# Process large transaction dataset in batches -batcher = DataBatcher(batch_size=5000, timeout_seconds=10.0) -total_processed = 0 -start_time = time.time() - -query = "MATCH (t:Transaction) WHERE t.amount > $min_amount RETURN t" -params = {"min_amount": 1000} - -for transaction in extensions.stream_query("transactions_graph", query, params): - batch = batcher.add(transaction) - - if batch: - # Process batch - process_transaction_batch(batch) - total_processed += len(batch) - - if total_processed % 50000 == 0: - elapsed = time.time() - start_time - print( - f"Processed {total_processed:,} transactions in {format_duration(int(elapsed * 1000))}" - ) - -# Process final batch -final_batch = batcher.flush() -if final_batch: - process_transaction_batch(final_batch) - total_processed += len(final_batch) - -print(f"Processed {total_processed:,} transactions total") -``` - -### Error Handling - -```python -from robosystems_client.clients import QueuedQueryError +from robosystems_client.clients import OperatorQueryRequest, OperatorRunError +request = OperatorQueryRequest(message="What changed in revenue?", thread_id=thread_id) try: - result = extensions.execute_query("graph_id", "COMPLEX LONG RUNNING QUERY") - print(f"Query completed: {len(result.data)} results") - -except QueuedQueryError as e: - print(f"Query was queued at position {e.queue_info.queue_position}") - print(f"Estimated wait time: {e.queue_info.estimated_wait_seconds}s") - - # Wait for completion or cancel - choice = input("Wait for completion? (y/n): ") - if choice.lower() == "y": - result = extensions.monitor_operation(e.queue_info.operation_id) - print(f"Query completed: {result.status}") - else: - cancelled = extensions.cancel_operation(e.queue_info.operation_id) - print(f"Query cancelled: {cancelled}") - -except Exception as e: - print(f"Query failed: {e}") + result = clients.operator.execute_query(graph_id, request) +except OperatorRunError as run: + print(run.status, run.message, run.writes) # writes made before the run stopped ``` -## Performance Optimization +Reuse a `thread_id` of your choosing, such as a UUID, for follow-up questions. A queued run is followed over SSE, then over `/v1/operations/{id}/status` if the stream gives no verdict; `OperatorOptions(poll_interval=5)`, passed as the third argument, sets the seconds between polls. `OperatorOptions(max_wait=0)` raises `QueuedOperatorError` instead of waiting. -### Connection Pooling +## Ledger -Extensions automatically manage SSE connections with pooling: +`LedgerClient` reads through GraphQL at `POST /extensions/{graph_id}/graphql` and writes through `POST /extensions/roboledger/{graph_id}/operations/{op}`. Its API methods take `graph_id` first. Reads return typed models, or `None` when the record is missing. Writes return the operation's result, not its envelope. Many methods take `entity_id`; omitted, they act on the group parent. ```python -# Configure connection limits -config = RoboSystemsClientConfig( - max_retries=5, - retry_delay=2000, # 2 seconds - timeout=60, +ledger = clients.ledger +entity = ledger.get_entity(graph_id) +calendar = ledger.get_fiscal_calendar(graph_id) +print(entity.name, calendar.closed_through, calendar.closeable_now) +accounts = ledger.list_accounts(graph_id, classification="asset", limit=50).accounts + +ledger.create_journal_entry( + graph_id, + posting_date="2026-03-31", + memo="March accrual", + line_items=[ + {"element_id": expense_element_id, "debit_amount": 50000, "credit_amount": 0}, + {"element_id": accrual_element_id, "debit_amount": 0, "credit_amount": 50000}, + ], + idempotency_key="accrual-2026-03", ) - -extensions = RoboSystemsClients(config) -``` - -### Query Optimization - -Use query analysis tools: - -```python -from robosystems_client.clients import estimate_query_cost, validate_cypher_query - -# Analyze before execution -query = "MATCH (c:Company) WHERE c.revenue > 1000000 RETURN c" - -# Check syntax -validation = validate_cypher_query(query) -if not validation["valid"]: - print("Query has syntax errors:", validation["issues"]) - -# Estimate cost -cost = estimate_query_cost(query) -print(f"Query complexity: {cost['complexity_category']}") - -# Follow recommendations -for rec in cost["recommendations"]: - print(f"Optimization tip: {rec}") - -# Execute only if reasonable complexity -if cost["complexity_category"] in ["low", "medium"]: - result = extensions.execute_query("graph_id", query) -else: - print("Query may be too expensive - consider optimization") +closed = ledger.close_period(graph_id, "2026-03", note="March close") +print(closed.period, closed.entries_posted) ``` -### Caching Strategy +Line amounts are in cents, and a journal entry is a draft unless you pass `status="posted"`. Only some writes take `idempotency_key`; check the signature. A selection of the rest: -```python -# Set up tiered caching -schema_cache = CacheManager(max_size=10, ttl_seconds=3600) # 1 hour for schemas -query_cache = CacheManager(max_size=100, ttl_seconds=300) # 5 minutes for queries -results_cache = CacheManager(max_size=1000, ttl_seconds=60) # 1 minute for results - -# Use appropriate cache based on query type -if "SCHEMA" in query.upper(): - cache = schema_cache -elif any(keyword in query.upper() for keyword in ["COUNT", "SUM", "AVG"]): - cache = query_cache -else: - cache = results_cache -``` +| Area | Methods | +| ---- | ------- | +| Setup and chart | `initialize_ledger`, `initialize_chart_of_accounts`, `create_entity`, `update_entity`, `link_bank_account`, `get_account_tree`, `list_elements` | +| Mapping | `list_taxonomies`, `list_mappings`, `create_mapping_association`, `auto_map_elements`, `list_unmapped_elements` | +| Events and entries | `create_event_block`, `update_event_block`, `preview_event_block`, `list_event_blocks`, `accept_suggestions`, `update_journal_entry`, `reverse_journal_entry` | +| Balances and views | `get_trial_balance`, `get_account_register`, `get_aged_balances`, `get_summary`, `build_fact_grid`, `financial_statement_analysis`, `live_financial_statement` | +| Schedules and blocks | `create_schedule`, `preview_schedule`, `rebuild_schedule`, `create_information_block`, `get_information_block` | +| Close and consolidation | `get_period_close_status`, `list_period_drafts`, `reopen_period`, `set_close_target`, `list_reconciliations`, `sign_off_reconciliation`, `record_acquisition`, `consolidation_worksheet` | +| Reports | `create_report`, `get_report`, `get_statement`, `file_report`, `share_report`, `download_report_bundle` | -## Testing - -Run the test suite: - -```bash -# Run all tests -python run_tests.py - -# Or individual test suites -python -c "from robosystems_client.clients.tests.test_unit import run_unit_tests; run_unit_tests()" -python -c "from robosystems_client.clients.tests.test_integration import run_integration_tests; run_integration_tests()" -``` +## Investor -### Writing Tests +`InvestorClient` follows the same pattern, with writes at `POST /extensions/roboinvestor/{graph_id}/operations/{op}`. Positions have no write methods of their own; they change through `create_portfolio_block` and `update_portfolio_block`. ```python -from robosystems_client.clients import AuthenticatedClients -from unittest.mock import Mock, patch - - -def test_query_execution(): - extensions = AuthenticatedClients("test-key") - - with patch("robosystems_client.clients.auth_integration.sync_detailed") as mock_query: - # Mock successful response - mock_response = Mock() - mock_response.parsed.data = [{"count": 100}] - mock_query.return_value = mock_response - - result = extensions.execute_cypher_query("test_graph", "MATCH (n) RETURN count(n)") - assert result["data"] == [{"count": 100}] +investor = clients.investor +body = {"name": "Example Common", "security_type": "common_stock"} +security = investor.create_security(graph_id, body) +positions = [{"security_id": security.id, "quantity": 100}] +body = {"portfolio": {"name": "Core"}, "positions": positions} +block = investor.create_portfolio_block(graph_id, body) +print(investor.list_portfolios(graph_id, limit=25).portfolios) +print(investor.get_holdings(graph_id, block.id).total_positions) ``` -## Configuration - -### Environment Variables +It also has `get_portfolio_block`, `delete_portfolio_block` (pass `confirm_active_positions=True` when positions are active), `list_securities`, `get_security`, `update_security`, `delete_security` (sets `is_active` to false), `list_positions` and `get_position`. -```bash -# API Configuration -export ROBOSYSTEMS_API_URL="https://api.robosystems.ai" -export ROBOSYSTEMS_API_KEY="your-api-key" -export ROBOSYSTEMS_GRAPH_ID="your-graph-id" - -# Connection Settings -export ROBOSYSTEMS_MAX_RETRIES="5" -export ROBOSYSTEMS_TIMEOUT="60" -export ROBOSYSTEMS_RETRY_DELAY="2000" -``` +## Library -### Configuration Object +`LibraryClient` is read-only. `graph_id` defaults to `LIBRARY_GRAPH_ID` (`"library"`), which reads the shared library. Pass a tenant `graph_id` to read that graph's copy plus its own extensions. A credential is still required. ```python -from robosystems_client.clients import RoboSystemsClientConfig - -config = RoboSystemsClientConfig( - base_url="https://api.robosystems.ai", - headers={"X-Custom-Header": "value", "X-Client-Version": "1.0.0"}, - max_retries=3, - retry_delay=1500, - timeout=45, -) - -extensions = RoboSystemsClients(config) +library = clients.library +print([(t.standard, t.version) for t in library.list_library_taxonomies()]) +print([e.qname for e in library.search_library_elements("revenue", limit=10)]) +print(library.get_library_element(qname="rs-gaap:ProceedsFromIssuanceOfLongTermDebt")) ``` -## API Reference - -### Core Classes - -- **`RoboSystemsClients`** - Main extensions class -- **`AuthenticatedClients`** - API key authentication -- **`CookieAuthClients`** - Cookie authentication -- **`TokenClients`** - JWT token authentication -- **`AsyncRoboSystemsClients`** - Async version - -### SSE Components - -- **`SSEClient`** - Server-Sent Events client -- **`SSEConfig`** - SSE configuration -- **`EventType`** - Standard event types enum - -### Query Components - -- **`QueryClient`** - Enhanced query execution -- **`QueryBuilder`** - Programmatic query building -- **`QueryResult`** - Query result data structure - -### Utilities +It also has `get_library_taxonomy`, `list_library_elements`, `list_library_taxonomy_arcs`, `list_library_structures`, `get_library_element_arcs`, `get_library_element_classifications` and `get_library_element_equivalents`. `LedgerClient`, `InvestorClient` and `LibraryClient` can also be built alone from a plain dict, such as `LibraryClient({"base_url": "https://api.robosystems.ai", "token": api_key})`. `base_url` is required, and `timeout` defaults to 60 this way. -- **`ResultProcessor`** - Format conversion utilities -- **`CacheManager`** - Smart caching with LRU eviction -- **`ProgressTracker`** - Operation progress monitoring -- **`DataBatcher`** - Batch processing helper +## Graphs, files, tables and documents -### Functions - -- **`validate_cypher_query(query)`** - Query syntax validation -- **`estimate_query_cost(query, params)`** - Complexity analysis -- **`format_duration(milliseconds)`** - Human-readable time formatting -- **`create_clients(method, **kwargs)`** - Extensions factory - -## Troubleshooting - -### Common Issues - -**Import Errors** ```python -# Make sure all dependencies are installed -pip install httpx # For SSE support -pip install pandas # For DataFrame conversion (optional) -``` +from robosystems_client.clients import GraphMetadata, InitialEntityData -**Authentication Failures** -```python -# Verify API key is valid -extensions = AuthenticatedClients("your-api-key") -try: - result = extensions.execute_query("graph_id", "MATCH (n) RETURN count(n) LIMIT 1") - print("Authentication successful") -except Exception as e: - print(f"Auth failed: {e}") -``` +metadata = GraphMetadata(graph_name="Example Holdings") +entity = InitialEntityData(name="Example Holdings LLC", uri="https://example.com") +graph_id = clients.graphs.create_graph_and_wait(metadata, entity, timeout=120) -**Connection Issues** -```python -# Check base URL and connectivity -import httpx +uploaded = clients.files.upload(graph_id, "Entity", "entity.parquet") +staged = clients.tables.query(graph_id, "SELECT * FROM Entity", limit=5) +done = clients.graphs.materialize(graph_id) +print(uploaded.success, staged.rows, done.total_rows) -try: - response = httpx.get("https://api.robosystems.ai/health", timeout=10) - print(f"API Status: {response.status_code}") -except Exception as e: - print(f"Connection failed: {e}") +clients.documents.upload(graph_id, title="Close memo", content="# March\n\nNotes.") +hits = clients.documents.search(graph_id, "close memo", size=5).hits ``` -**SSE Streaming Issues** -```python -# Enable debug logging for SSE events -config = SSEConfig(base_url="https://api.robosystems.ai") -client = SSEClient(config) - +`upload`, `tables.query` and `materialize` report failure through `success` and `error` on their results. `MaterializationOptions` sets `rebuild`, `force`, `on_progress` and `timeout`. `files.upload` reads the whole file into memory and sends it as Parquet. Its `row_count` is 0 when a large file stages asynchronously. `change_tier` takes `ladybug-standard`, `ladybug-large` or `ladybug-xlarge`. `delete_graph` cancels the subscription and starts deprovisioning, which `at_period_end=True` defers. -def debug_handler(data): - print(f"SSE Event: {data}") +Subgraphs and backups have no facade method. `AuthenticatedClients` and `TokenClients` expose `authenticated_client` for the generated functions in `robosystems_client.api`, such as `sync_detailed(graph_id, client=clients.authenticated_client, body=body)` from `robosystems_client.api.graph_operations.create_subgraph`. Calls through it do not replay `429` responses. +## Async -client.on("event", debug_handler) -``` - -### Debug Mode +`AsyncRoboSystemsClients` has only `query` and `operations`, plus `execute_query`, `stream_query`, `monitor_operation`, `get_operation_status`, `cancel_operation`, `create_sse_client` and `close`. There is no async ledger, investor, library, graph, file, table, document or operator client, and no auth subclass. Its `base_url` defaults to localhost, its REST calls do not replay `429` responses, and its `stream_query` also collects the result first. ```python -# Enable detailed logging -import logging +from robosystems_client.clients import AsyncRoboSystemsClients, RoboSystemsClientConfig -logging.basicConfig(level=logging.DEBUG) - -# This will show internal HTTP requests and SSE events -extensions = AuthenticatedClients("your-key") +headers = {"X-API-Key": api_key} +config = RoboSystemsClientConfig(base_url="https://api.robosystems.ai", headers=headers) +clients = AsyncRoboSystemsClients(config) +result = await clients.execute_query("sec", "MATCH (e:Entity) RETURN count(e) AS total") +await clients.close() ``` -## Contributing - -1. Fork the repository -2. Create a feature branch -3. Add tests for new functionality -4. Run the test suite: `python run_tests.py` -5. Submit a pull request - -## Support - -- **API Reference**: [api.robosystems.ai](https://api.robosystems.ai) -- **Issues**: [GitHub Issues](https://github.com/RoboFinSystems/robosystems-python-client/issues) +## Errors and retries ---- +| Call | On failure | +| ---- | ---------- | +| Ledger, investor and library reads | `GraphQLError`, with `errors` and `status_code` | +| Ledger and investor writes | `RuntimeError` | +| `query`, `operator`, `documents` | `Exception`; `documents.get` and `get_section` return `None` and `documents.delete` returns `False` for a 404 | +| `files.upload`, `tables.query`, `graphs.materialize` | A result with `success=False` and `error` set | +| `files.list`, `files.get`, `files.delete`, `tables.list` | `[]`, `None` or `False`, with the cause only in the log | +| Other `graphs` methods | Raise, usually `RuntimeError`, `ValueError` or `TimeoutError` | +| `operations.get_operation_status` | A dict with `status` of `"error"` or `"unknown"` | -**RoboSystems Python Client Extensions** - Production-ready streaming, monitoring, and query capabilities for financial knowledge graphs. +The synchronous facades replay `429` responses with exponential backoff and jitter, capped by `Retry-After`. No other HTTP status is replayed. There is no shared connection pool: most calls build a new HTTP client. To check that the API is reachable, call `GET /v1/status`.