From b574fe398318cd1a6d287c17d3ba5dbe7fae32cb Mon Sep 17 00:00:00 2001 From: Hector Jacinto Murillo Date: Mon, 3 Aug 2026 09:44:09 +0200 Subject: [PATCH 1/6] doc: update 'doc' folder content --- docs/csm-data/store/load-parquet-folder.md | 11 + docs/csm-data/store/output.md | 11 + docs/tutorials/cosmotech-api.md | 264 ++++++++++----------- docs/tutorials/csm-data.md | 5 +- docs/tutorials/datastore.md | 3 +- 5 files changed, 151 insertions(+), 143 deletions(-) create mode 100644 docs/csm-data/store/load-parquet-folder.md create mode 100644 docs/csm-data/store/output.md diff --git a/docs/csm-data/store/load-parquet-folder.md b/docs/csm-data/store/load-parquet-folder.md new file mode 100644 index 00000000..ae964ecf --- /dev/null +++ b/docs/csm-data/store/load-parquet-folder.md @@ -0,0 +1,11 @@ +--- +hide: + - toc +description: "Command help: `csm-data store load-parquet-folder`" +--- +# load-parquet-folder + +!!! info "Help command" + ```text + --8<-- "generated/commands_help/csm-data/store/load-parquet-folder.txt" + ``` diff --git a/docs/csm-data/store/output.md b/docs/csm-data/store/output.md new file mode 100644 index 00000000..1262fa71 --- /dev/null +++ b/docs/csm-data/store/output.md @@ -0,0 +1,11 @@ +--- +hide: + - toc +description: "Command help: `csm-data store output`" +--- +# output + +!!! info "Help command" + ```text + --8<-- "generated/commands_help/csm-data/store/output.txt" + ``` diff --git a/docs/tutorials/cosmotech-api.md b/docs/tutorials/cosmotech-api.md index f5935e0c..8f186aba 100644 --- a/docs/tutorials/cosmotech-api.md +++ b/docs/tutorials/cosmotech-api.md @@ -1,5 +1,5 @@ --- -description: "Comprehensive guide to working with the CosmoTech API in CoAL: authentication, workspaces, Twin Data Layer, and more" +description: "Comprehensive guide to working with the CosmoTech API in CoAL: authentication, workspaces, runners, and datasets" --- # Working with the CosmoTech API @@ -7,8 +7,8 @@ description: "Comprehensive guide to working with the CosmoTech API in CoAL: aut !!! abstract "Objective" + Understand how to authenticate and connect to the CosmoTech API + Learn to work with workspaces for file management - + Master the Twin Data Layer for graph data operations + Implement runner and run data management + + Upload and download datasets + Build complete workflows integrating multiple API features ## Introduction to the CosmoTech API Integration @@ -17,17 +17,21 @@ The CosmoTech Acceleration Library (CoAL) provides a comprehensive set of tools - Authenticate with different identity providers - Manage workspaces and files -- Work with the Twin Data Layer for graph data - Handle runners and runs +- Upload and download datasets - Process and transform data - Build end-to-end workflows -The API integration is organized into several modules, each focused on specific functionality: +The API integration is organized into two sub-packages under `cosmotech.coal.cosmotech_api`: -- **connection**: Authentication and API client management -- **workspace**: Workspace file operations -- **twin_data_layer**: Graph data management -- **runner**: Runner and run data operations +- **`objects/`**: Core building blocks + - `connection` — `Connection` class: authentication and `ApiClient` management + - `parameters` — `Parameters` class: typed access to runner parameters +- **`apis/`**: High-level wrappers for each CosmoTech API resource + - `DatasetApi` — dataset upload, download, and parts management + - `RunnerApi` — runner metadata and data download + - `WorkspaceApi` — workspace file listing, download, and upload + - `RunApi`, `OrganizationApi`, `SolutionApi`, `MetaApi` — additional resource wrappers !!! info "API vs CLI" While the `csm-data` CLI provides command-line tools for many common operations, the direct API integration offers more flexibility and programmatic control. Use the API integration when you need to: @@ -45,10 +49,24 @@ The first step in working with the CosmoTech API is establishing a connection. C - Azure Entra (formerly Azure AD) authentication - Keycloak authentication -The `get_api_client()` function automatically detects which authentication method to use based on the environment variables you've set. +The `Connection` class automatically detects which authentication method to use based on the environment variables present. ```python title="Basic connection setup" linenums="1" ---8<-- 'tutorial/cosmotech-api/connection_setup.py' +from cosmotech.coal.cosmotech_api.objects.connection import Connection + +# Connection auto-detects authentication from environment variables +connection = Connection() +api_client = connection.api_client # cosmotech_api.ApiClient +``` + +All API wrapper classes (`WorkspaceApi`, `RunnerApi`, `DatasetApi`, …) extend `Connection` and set themselves up automatically — you do not need to create the `Connection` separately unless you want direct access to the raw `ApiClient`. + +```python +from cosmotech.coal.cosmotech_api.apis import WorkspaceApi, RunnerApi, DatasetApi + +ws_api = WorkspaceApi() # auth resolved automatically +runner_api = RunnerApi() +dataset_api = DatasetApi() ``` !!! tip "Environment Variables" @@ -86,54 +104,47 @@ Keycloak authentication requires these environment variables: ## Working with Workspaces -Workspaces in the CosmoTech platform provide a way to organize and share files. The CoAL library offers functions for listing, downloading, and uploading files in workspaces. +Workspaces in the CosmoTech platform provide a way to organize and share files. `WorkspaceApi` offers methods for listing, downloading, and uploading files. ```python title="Workspace operations" linenums="1" ---8<-- 'tutorial/cosmotech-api/workspace_operations.py' -``` +from pathlib import Path +from cosmotech.coal.cosmotech_api.apis import WorkspaceApi -### Listing Files +ws_api = WorkspaceApi() -The `list_workspace_files` function allows you to list files in a workspace with a specific prefix: +# List files whose names start with a given prefix +files = ws_api.list_filtered_workspace_files( + organization_id, workspace_id, file_prefix="inputs/" +) -```python -files = list_workspace_files(api_client, organization_id, workspace_id, file_prefix) -``` +# Download a file to a local directory +local_path = ws_api.download_workspace_file( + organization_id, workspace_id, + file_name="inputs/data.csv", + target_dir=Path("/tmp/downloads"), +) -This is useful for finding files in a specific directory or with a specific naming pattern. +# Upload a local file to the workspace +uploaded_name = ws_api.upload_workspace_file( + organization_id, workspace_id, + file_path="/tmp/results/output.csv", + workspace_path="outputs/", # trailing slash → preserves original filename + overwrite=True, +) +``` -### Downloading Files +### Listing Files -The `download_workspace_file` function downloads a file from the workspace to a local directory: +`list_filtered_workspace_files` returns all workspace files whose `file_name` starts with the given prefix. It raises `ValueError` when no matching files are found. -```python -downloaded_file = download_workspace_file( - api_client, - organization_id, - workspace_id, - file_to_download, - target_directory -) -``` +### Downloading Files -If the file is in a subdirectory in the workspace, the function will create the necessary local subdirectories. +`download_workspace_file` writes the file content to `target_dir / file_name`, creating any necessary intermediate directories. ### Uploading Files -The `upload_workspace_file` function uploads a local file to the workspace: +`upload_workspace_file` uploads a single local file. The `workspace_path` parameter can be: -```python -uploaded_file = upload_workspace_file( - api_client, - organization_id, - workspace_id, - file_to_upload, - workspace_destination, - overwrite=True -) -``` - -The `workspace_destination` parameter can be: - A specific file path in the workspace - A directory path ending with `/`, in which case the original filename is preserved @@ -144,133 +155,105 @@ The `workspace_destination` parameter can be: - End directory paths with a trailing slash (`/`) - Use relative paths from the workspace root -## Twin Data Layer Operations - -The Twin Data Layer (TDL) is a graph database that stores nodes and relationships. CoAL provides tools for working with the TDL, particularly for preparing and sending CSV data. - -```python title="Twin Data Layer operations" linenums="1" ---8<-- 'tutorial/cosmotech-api/twin_data_layer.py' -``` - -### CSV File Format - -The TDL expects CSV files in a specific format: - -- **Node files**: Must have an `id` column and can have additional property columns -- **Relationship files**: Must have `src` and `dest` columns and can have additional property columns - -The filename (without the `.csv` extension) becomes the node label or relationship type in the graph. - -### Parsing CSV Files - -The `CSVSourceFile` class helps parse CSV files and determine if they represent nodes or relationships: - -```python -csv_file = CSVSourceFile(file_path) -print(f"Is node: {csv_file.is_node}") -print(f"Fields: {csv_file.fields}") -``` +## Dataset Management -### Generating Cypher Queries +`DatasetApi` provides helpers for uploading datasets and managing their parts (files that compose the dataset). -The `generate_query_insert` method creates Cypher queries for inserting data into the TDL: +```python title="Dataset upload" linenums="1" +from cosmotech.coal.cosmotech_api.apis import DatasetApi -```python -query = csv_file.generate_query_insert() -``` +dataset_api = DatasetApi() -These queries can then be executed using the TwinGraphApi: +# Upload a single file as a dataset +dataset_api.upload_dataset( + organization_id=organization_id, + dataset_id=dataset_id, + file_path="/tmp/data/customers.csv", +) -```python -twin_graph_api.run_twin_graph_cypher_query( +# Upload multiple parts from a folder (one part per file) +dataset_api.upload_dataset_parts( organization_id=organization_id, - workspace_id=workspace_id, - twin_graph_id=twin_graph_id, - twin_graph_cypher_query={ - "query": query, - "parameters": params - } + dataset_id=dataset_id, + folder_path="/tmp/data/parts/", +) + +# Download a dataset to a local directory +dataset_api.download_dataset( + dataset_id=dataset_id, ) ``` -!!! warning "Node References" - When creating relationships, make sure the nodes referenced by the `src` and `dest` columns already exist in the graph. Otherwise, the relationship creation will fail. +!!! info "Dataset Parts" + When uploading parts, the part name is derived from the filename without its extension. ## Runner and Run Management -Runners and runs are central concepts in the CosmoTech platform. CoAL provides functions for working with runner data, parameters, and associated datasets. +Runners and runs are central concepts in the CosmoTech platform. `RunnerApi` provides methods for retrieving runner metadata and downloading all associated data (parameters and datasets). ```python title="Runner operations" linenums="1" ---8<-- 'tutorial/cosmotech-api/runner_operations.py' -``` - -### Getting Runner Data - -The `get_runner_data` function retrieves information about a runner: - -```python -runner_data = get_runner_data(organization_id, workspace_id, runner_id) -``` - -### Working with Parameters - -The `get_runner_parameters` function extracts parameters from runner data: +from cosmotech.coal.cosmotech_api.apis import RunnerApi -```python -parameters = get_runner_parameters(runner_data) -``` - -### Downloading Runner Data +runner_api = RunnerApi() -The `download_runner_data` function downloads all data associated with a runner, including parameters and datasets: - -```python -result = download_runner_data( - organization_id=organization_id, - workspace_id=workspace_id, +# Retrieve runner metadata as a dict +metadata = runner_api.get_runner_metadata( runner_id=runner_id, - parameter_folder=str(param_dir), - dataset_folder=str(dataset_dir), - write_json=True, - write_csv=True, - fetch_dataset=True, + # optionally scope returned fields: + # include=["parametersValues", "datasetList"] ) -``` - -This function: -- Downloads parameters and writes them as JSON and/or CSV files -- Downloads associated datasets -- Organizes everything in the specified directories - -!!! tip "Dataset References" - Runners can reference datasets in two ways: - - Through parameters with the `%DATASETID%` variable type - - Through the `dataset_list` property - - The `download_runner_data` function handles both types of references. +# Download runner parameters and datasets +runner_api.download_runner_data( + download_datasets="all", # or None to skip dataset download +) +``` ## Complete Workflow Example -Putting it all together, here's a complete workflow that demonstrates how to use the CosmoTech API for a data processing pipeline: +Putting it all together, here's a typical end-to-end workflow for a CosmoTech data processing pipeline: ```python title="Complete workflow" linenums="1" ---8<-- 'tutorial/cosmotech-api/complete_workflow.py' +from cosmotech.coal.cosmotech_api.apis import RunnerApi, WorkspaceApi, DatasetApi +from pathlib import Path + +# 1. Download runner parameters and datasets +runner_api = RunnerApi() +runner_api.download_runner_data(download_datasets="all") + +# 2. Process the data (application-specific logic) +# ... + +# 3. Upload results back to the workspace +ws_api = WorkspaceApi() +ws_api.upload_workspace_file( + organization_id, workspace_id, + file_path="/tmp/results/report.csv", + workspace_path="outputs/", + overwrite=True, +) + +# 4. Update a dataset with processed parts +dataset_api = DatasetApi() +dataset_api.upload_dataset_parts( + organization_id=organization_id, + dataset_id=output_dataset_id, + folder_path="/tmp/results/parts/", +) ``` This workflow: -1. Downloads runner data (parameters and datasets) -2. Processes the data (calculates loyalty scores for customers) -3. Uploads the processed data to the workspace -4. Prepares the data for the Twin Data Layer -5. Generates a report with statistics and insights +1. Downloads runner parameters and associated datasets +2. Processes the data (application-specific logic) +3. Uploads processed results to the workspace +4. Updates a dataset with the processed output parts !!! tip "Real-world Workflows" In real-world scenarios, you might: - Use more complex data transformations - - Integrate with external systems + - Integrate with other Python code or services - Implement error handling and retries - Add logging and monitoring - Parallelize operations for better performance @@ -286,6 +269,8 @@ This workflow: ### Error Handling ```python +import cosmotech_api + try: # API operations except cosmotech_api.exceptions.ApiException as e: @@ -294,9 +279,6 @@ except cosmotech_api.exceptions.ApiException as e: except Exception as e: # Handle other errors print(f"Error: {e}") -finally: - # Always close the client - api_client.close() ``` ### Performance Considerations diff --git a/docs/tutorials/csm-data.md b/docs/tutorials/csm-data.md index d31535bc..2c726706 100644 --- a/docs/tutorials/csm-data.md +++ b/docs/tutorials/csm-data.md @@ -17,7 +17,7 @@ description: "Comprehensive guide to the csm-data CLI: a powerful data managemen The CLI is organized into several command groups, each focused on specific types of data operations: - **api**: Commands for interacting with the CosmoTech API -- **store**: Commands for working with the CoAL datastore +- **store**: Commands for working with the CoAL datastore (load CSV/Parquet, dump, output channels, delete, reset) - **s3-bucket-***: Commands for S3 bucket operations (download, upload, delete) - **adx-send-runnerdata**: Command for sending runner data to Azure Data Explorer - **az-storage-upload**: Command for uploading to Azure Storage @@ -132,8 +132,11 @@ The `store` command group provides tools for working with the CoAL datastore: These commands allow you to: - Load data from CSV files into the datastore +- Load data from Parquet files into the datastore - Dump datastore contents to various destinations (S3, Azure, PostgreSQL) +- Send the full datastore output through one or more configured output channels - List tables in the datastore +- Delete specific tables from the datastore - Reset the datastore ## Common Workflows and Integration Patterns diff --git a/docs/tutorials/datastore.md b/docs/tutorials/datastore.md index 9135e688..6d7d1b1d 100644 --- a/docs/tutorials/datastore.md +++ b/docs/tutorials/datastore.md @@ -17,7 +17,7 @@ The datastore is a powerful data management abstraction that provides a unified The core idea behind the datastore is to provide a robust, flexible system for data management that simplifies working with different data formats while offering persistence and advanced query capabilities. !!! info "Key Features" - - Format flexibility (Python dictionaries, CSV files, Pandas DataFrames, PyArrow Tables) + - Format flexibility (Python dictionaries, CSV files, Pandas DataFrames, PyArrow Tables, Parquet files) - Persistent storage in SQLite - SQL query capabilities - Simplified data pipeline management @@ -32,6 +32,7 @@ The datastore works seamlessly with multiple data formats: - CSV files - Pandas DataFrames - PyArrow Tables +- Parquet files This flexibility eliminates the need for manual format conversions and allows you to work with data in your preferred format. From 7d9c784458ac35a0adde4fb7528c280fdf5c8fe7 Mon Sep 17 00:00:00 2001 From: Hector Jacinto Murillo Date: Mon, 3 Aug 2026 10:10:33 +0200 Subject: [PATCH 2/6] doc: update 'tutorial' content --- docs/tutorials/cosmotech-api.md | 74 +--- docs/tutorials/datastore.md | 6 + tutorial/cosmotech-api/complete_workflow.py | 319 ++++++------------ tutorial/cosmotech-api/connection_setup.py | 74 ++-- tutorial/cosmotech-api/runner_operations.py | 128 ++----- tutorial/cosmotech-api/twin_data_layer.py | 192 ++++------- .../cosmotech-api/workspace_operations.py | 87 ++--- tutorial/csm-data/api_env_variables.bash | 2 +- tutorial/csm-data/complete_pipeline.bash | 4 +- tutorial/csm-data/csm_orc_integration.json | 10 +- tutorial/csm-data/run_load_data.bash | 5 +- tutorial/datastore/parquet_files.py | 51 +++ 12 files changed, 323 insertions(+), 629 deletions(-) create mode 100644 tutorial/datastore/parquet_files.py diff --git a/docs/tutorials/cosmotech-api.md b/docs/tutorials/cosmotech-api.md index 8f186aba..71d1257e 100644 --- a/docs/tutorials/cosmotech-api.md +++ b/docs/tutorials/cosmotech-api.md @@ -52,11 +52,7 @@ The first step in working with the CosmoTech API is establishing a connection. C The `Connection` class automatically detects which authentication method to use based on the environment variables present. ```python title="Basic connection setup" linenums="1" -from cosmotech.coal.cosmotech_api.objects.connection import Connection - -# Connection auto-detects authentication from environment variables -connection = Connection() -api_client = connection.api_client # cosmotech_api.ApiClient +--8<-- 'tutorial/cosmotech-api/connection_setup.py' ``` All API wrapper classes (`WorkspaceApi`, `RunnerApi`, `DatasetApi`, …) extend `Connection` and set themselves up automatically — you do not need to create the `Connection` separately unless you want direct access to the raw `ApiClient`. @@ -107,30 +103,7 @@ Keycloak authentication requires these environment variables: Workspaces in the CosmoTech platform provide a way to organize and share files. `WorkspaceApi` offers methods for listing, downloading, and uploading files. ```python title="Workspace operations" linenums="1" -from pathlib import Path -from cosmotech.coal.cosmotech_api.apis import WorkspaceApi - -ws_api = WorkspaceApi() - -# List files whose names start with a given prefix -files = ws_api.list_filtered_workspace_files( - organization_id, workspace_id, file_prefix="inputs/" -) - -# Download a file to a local directory -local_path = ws_api.download_workspace_file( - organization_id, workspace_id, - file_name="inputs/data.csv", - target_dir=Path("/tmp/downloads"), -) - -# Upload a local file to the workspace -uploaded_name = ws_api.upload_workspace_file( - organization_id, workspace_id, - file_path="/tmp/results/output.csv", - workspace_path="outputs/", # trailing slash → preserves original filename - overwrite=True, -) +--8<-- 'tutorial/cosmotech-api/workspace_operations.py' ``` ### Listing Files @@ -192,21 +165,7 @@ dataset_api.download_dataset( Runners and runs are central concepts in the CosmoTech platform. `RunnerApi` provides methods for retrieving runner metadata and downloading all associated data (parameters and datasets). ```python title="Runner operations" linenums="1" -from cosmotech.coal.cosmotech_api.apis import RunnerApi - -runner_api = RunnerApi() - -# Retrieve runner metadata as a dict -metadata = runner_api.get_runner_metadata( - runner_id=runner_id, - # optionally scope returned fields: - # include=["parametersValues", "datasetList"] -) - -# Download runner parameters and datasets -runner_api.download_runner_data( - download_datasets="all", # or None to skip dataset download -) +--8<-- 'tutorial/cosmotech-api/runner_operations.py' ``` ## Complete Workflow Example @@ -214,32 +173,7 @@ runner_api.download_runner_data( Putting it all together, here's a typical end-to-end workflow for a CosmoTech data processing pipeline: ```python title="Complete workflow" linenums="1" -from cosmotech.coal.cosmotech_api.apis import RunnerApi, WorkspaceApi, DatasetApi -from pathlib import Path - -# 1. Download runner parameters and datasets -runner_api = RunnerApi() -runner_api.download_runner_data(download_datasets="all") - -# 2. Process the data (application-specific logic) -# ... - -# 3. Upload results back to the workspace -ws_api = WorkspaceApi() -ws_api.upload_workspace_file( - organization_id, workspace_id, - file_path="/tmp/results/report.csv", - workspace_path="outputs/", - overwrite=True, -) - -# 4. Update a dataset with processed parts -dataset_api = DatasetApi() -dataset_api.upload_dataset_parts( - organization_id=organization_id, - dataset_id=output_dataset_id, - folder_path="/tmp/results/parts/", -) +--8<-- 'tutorial/cosmotech-api/complete_workflow.py' ``` This workflow: diff --git a/docs/tutorials/datastore.md b/docs/tutorials/datastore.md index 6d7d1b1d..f65cdadc 100644 --- a/docs/tutorials/datastore.md +++ b/docs/tutorials/datastore.md @@ -90,6 +90,12 @@ The datastore provides specialized adapters for working with various data format --8<-- 'tutorial/datastore/pyarrow_tables.py' ``` +### Parquet Files + +```python title="Loading and exporting Parquet files" linenums="1" +--8<-- 'tutorial/datastore/parquet_files.py' +``` + ## Advanced use cases ### Joining multiple tables diff --git a/tutorial/cosmotech-api/complete_workflow.py b/tutorial/cosmotech-api/complete_workflow.py index df7d432b..2cb7e72e 100644 --- a/tutorial/cosmotech-api/complete_workflow.py +++ b/tutorial/cosmotech-api/complete_workflow.py @@ -4,245 +4,116 @@ import os import pathlib -from cosmotech_api.api.dataset_api import DatasetApi -from cosmotech_api.api.twin_graph_api import TwinGraphApi - -from cosmotech.coal.cosmotech_api.connection import get_api_client -from cosmotech.coal.cosmotech_api.runner import ( - download_runner_data, - get_runner_data, -) -from cosmotech.coal.cosmotech_api.twin_data_layer import CSVSourceFile -from cosmotech.coal.cosmotech_api.workspace import ( - download_workspace_file, - list_workspace_files, - upload_workspace_file, -) +from cosmotech.coal.cosmotech_api.apis import DatasetApi, RunnerApi, WorkspaceApi +from cosmotech.coal.utils.configuration import Configuration from cosmotech.coal.utils.logger import LOGGER -# Set up environment variables for authentication os.environ["CSM_API_URL"] = "https://api.cosmotech.com" # Replace with your API URL os.environ["CSM_API_KEY"] = "your-api-key" # Replace with your actual API key -# Organization, workspace, and runner IDs -organization_id = "your-organization-id" # Replace with your organization ID -workspace_id = "your-workspace-id" # Replace with your workspace ID -runner_id = "your-runner-id" # Replace with your runner ID -twin_graph_id = "your-twin-graph-id" # Replace with your twin graph ID +organization_id = "your-organization-id" +workspace_id = "your-workspace-id" +runner_id = "your-runner-id" +output_dataset_id = "your-output-dataset-id" -# Create directories for our workflow workflow_dir = pathlib.Path("./workflow_example") -workflow_dir.mkdir(exist_ok=True, parents=True) - input_dir = workflow_dir / "input" processed_dir = workflow_dir / "processed" output_dir = workflow_dir / "output" +for d in (input_dir, processed_dir, output_dir): + d.mkdir(exist_ok=True, parents=True) + +# Build a Configuration scoped to this runner +config = Configuration() +config.cosmotech.organization_id = organization_id +config.cosmotech.workspace_id = workspace_id +config.cosmotech.runner_id = runner_id +config.cosmotech.parameters_absolute_path = str(input_dir / "parameters") +config.cosmotech.dataset_absolute_path = str(input_dir / "datasets") +pathlib.Path(config.cosmotech.parameters_absolute_path).mkdir(exist_ok=True, parents=True) +pathlib.Path(config.cosmotech.dataset_absolute_path).mkdir(exist_ok=True, parents=True) + +# Step 1: Download runner parameters and datasets +print("\n=== Step 1: Download Runner Data ===") +runner_api = RunnerApi(config) +runner_api.download_runner_data(download_datasets=True) +print(f"Runner data downloaded to {input_dir}") + +# Step 2: Process the data +print("\n=== Step 2: Process Data ===") + +customers_file = pathlib.Path(config.cosmotech.dataset_absolute_path) / "customers.csv" +if not customers_file.exists(): + print("Creating sample customers.csv for demonstration") + with open(customers_file, "w", newline="") as f: + writer = csv.writer(f) + writer.writerow(["id", "name", "age", "city", "spending"]) + writer.writerow(["c1", "Alice", "30", "New York", "1500"]) + writer.writerow(["c2", "Bob", "25", "San Francisco", "2000"]) + writer.writerow(["c3", "Charlie", "35", "Chicago", "1200"]) + +customers = [] +with open(customers_file, "r") as f: + for row in csv.DictReader(f): + row["loyalty_score"] = str( + round(int(row["spending"]) / 100 + (int(row["age"]) - 20) / 10, 1) + ) + customers.append(row) -input_dir.mkdir(exist_ok=True, parents=True) -processed_dir.mkdir(exist_ok=True, parents=True) -output_dir.mkdir(exist_ok=True, parents=True) +processed_file = processed_dir / "customers_with_loyalty.csv" +with open(processed_file, "w", newline="") as f: + writer = csv.DictWriter(f, fieldnames=list(customers[0].keys())) + writer.writeheader() + writer.writerows(customers) -# Get the API client -api_client, connection_type = get_api_client() -LOGGER.info(f"Connected using: {connection_type}") +print(f"Processed data written to {processed_file}") +# Step 3: Upload the processed file to the workspace +print("\n=== Step 3: Upload Processed Data to Workspace ===") try: - # Step 1: Download runner data (parameters and datasets) - print("\n=== Step 1: Download Runner Data ===") - - runner_data = get_runner_data(organization_id, workspace_id, runner_id) - print(f"Runner name: {runner_data.name}") + ws_api = WorkspaceApi(config) + uploaded_name = ws_api.upload_workspace_file( + organization_id, + workspace_id, + str(processed_file), + "processed_data/", + overwrite=True, + ) + print(f"Uploaded as: {uploaded_name}") +except Exception as e: + print(f"Error uploading file: {e}") - result = download_runner_data( +# Step 4: Update a dataset with the processed output +print("\n=== Step 4: Update Output Dataset ===") +try: + dataset_api = DatasetApi(config) + dataset_api.upload_dataset( organization_id=organization_id, - workspace_id=workspace_id, - runner_id=runner_id, - parameter_folder=str(input_dir / "parameters"), - dataset_folder=str(input_dir / "datasets"), - write_json=True, - write_csv=True, + dataset_id=output_dataset_id, + file_path=str(processed_file), ) + print(f"Dataset {output_dataset_id} updated") +except Exception as e: + print(f"Error updating dataset: {e}") + +# Step 5: Generate a summary report +print("\n=== Step 5: Generate Report ===") +avg_loyalty = sum(float(c["loyalty_score"]) for c in customers) / len(customers) +report = { + "runner_id": runner_id, + "statistics": { + "total_customers": len(customers), + "average_loyalty_score": round(avg_loyalty, 1), + }, + "top_customers": sorted(customers, key=lambda c: float(c["loyalty_score"]), reverse=True)[:2], +} + +report_file = output_dir / "customer_report.json" +with open(report_file, "w") as f: + json.dump(report, f, indent=2) + +print(f"Report saved to {report_file}") +print(f"Total customers: {report['statistics']['total_customers']}") +print(f"Avg loyalty score: {report['statistics']['average_loyalty_score']}") +print("\nWorkflow completed successfully!") - print(f"Downloaded runner data to {input_dir}") - - # Step 2: Process the data - print("\n=== Step 2: Process Data ===") - - # For this example, we'll create a simple transformation: - # - Read a CSV file from the input - # - Transform it - # - Write the result to the processed directory - - # Let's assume we have a "customers.csv" file in the input directory - customers_file = input_dir / "datasets" / "customers.csv" - - # If the file doesn't exist for this example, create a sample one - if not customers_file.exists(): - print("Creating sample customers.csv file for demonstration") - customers_file.parent.mkdir(exist_ok=True, parents=True) - with open(customers_file, "w", newline="") as f: - writer = csv.writer(f) - writer.writerow(["id", "name", "age", "city", "spending"]) - writer.writerow(["c1", "Alice", "30", "New York", "1500"]) - writer.writerow(["c2", "Bob", "25", "San Francisco", "2000"]) - writer.writerow(["c3", "Charlie", "35", "Chicago", "1200"]) - - # Read the customers data - customers = [] - with open(customers_file, "r") as f: - reader = csv.DictReader(f) - for row in reader: - customers.append(row) - - print(f"Read {len(customers)} customers from {customers_file}") - - # Process the data: calculate a loyalty score based on age and spending - for customer in customers: - age = int(customer["age"]) - spending = int(customer["spending"]) - - # Simple formula: loyalty score = spending / 100 + (age - 20) / 10 - loyalty_score = round(spending / 100 + (age - 20) / 10, 1) - customer["loyalty_score"] = str(loyalty_score) - - # Write the processed data - processed_file = processed_dir / "customers_with_loyalty.csv" - with open(processed_file, "w", newline="") as f: - fieldnames = ["id", "name", "age", "city", "spending", "loyalty_score"] - writer = csv.DictWriter(f, fieldnames=fieldnames) - writer.writeheader() - writer.writerows(customers) - - print(f"Processed data written to {processed_file}") - - # Step 3: Upload the processed file to the workspace - print("\n=== Step 3: Upload Processed Data to Workspace ===") - - try: - uploaded_file = upload_workspace_file( - api_client, - organization_id, - workspace_id, - str(processed_file), - "processed_data/", # Destination in the workspace - overwrite=True, - ) - print(f"Uploaded processed file as: {uploaded_file}") - except Exception as e: - print(f"Error uploading file: {e}") - - # Step 4: Create a dataset from the processed data - print("\n=== Step 4: Create Dataset from Processed Data ===") - - # This step would typically involve: - # 1. Creating a dataset in the CosmoTech API - # 2. Uploading files to the dataset - - """ - # Create a dataset - dataset_api = DatasetApi(api_client) - - new_dataset = { - "name": "Customers with Loyalty Scores", - "description": "Processed customer data with calculated loyalty scores", - "tags": ["processed", "customers", "loyalty"] - } - - try: - dataset = dataset_api.create_dataset( - organization_id=organization_id, - workspace_id=workspace_id, - dataset=new_dataset - ) - - dataset_id = dataset.id - print(f"Created dataset with ID: {dataset_id}") - - # Upload the processed file to the dataset - # This would typically involve additional API calls - # ... - - except Exception as e: - print(f"Error creating dataset: {e}") - """ - - # Step 5: Send data to the Twin Data Layer - print("\n=== Step 5: Send Data to Twin Data Layer ===") - - # Parse the processed CSV file for the Twin Data Layer - customer_csv = CSVSourceFile(processed_file) - - # Generate a Cypher query for creating nodes - customer_query = customer_csv.generate_query_insert() - print(f"Generated Cypher query for Customer nodes:") - print(customer_query) - - # In a real scenario, you would send this data to the Twin Data Layer - """ - twin_graph_api = TwinGraphApi(api_client) - - # For each customer, create a node in the Twin Data Layer - with open(processed_file, "r") as f: - reader = csv.DictReader(f) - for row in reader: - # Create parameters for the Cypher query - params = {k: v for k, v in row.items()} - - # Execute the query - twin_graph_api.run_twin_graph_cypher_query( - organization_id=organization_id, - workspace_id=workspace_id, - twin_graph_id=twin_graph_id, - twin_graph_cypher_query={ - "query": customer_query, - "parameters": params - } - ) - """ - - # Step 6: Generate a report - print("\n=== Step 6: Generate Report ===") - - # Calculate some statistics - total_customers = len(customers) - avg_age = sum(int(c["age"]) for c in customers) / total_customers - avg_spending = sum(int(c["spending"]) for c in customers) / total_customers - avg_loyalty = sum(float(c["loyalty_score"]) for c in customers) / total_customers - - # Create a report - report = { - "report_date": "2025-02-28", - "runner_id": runner_id, - "statistics": { - "total_customers": total_customers, - "average_age": round(avg_age, 1), - "average_spending": round(avg_spending, 2), - "average_loyalty_score": round(avg_loyalty, 1), - }, - "top_customers": sorted(customers, key=lambda c: float(c["loyalty_score"]), reverse=True)[ - :2 - ], # Top 2 customers by loyalty score - } - - # Write the report to a JSON file - report_file = output_dir / "customer_report.json" - with open(report_file, "w") as f: - json.dump(report, f, indent=2) - - print(f"Report generated and saved to {report_file}") - - # Print a summary of the report - print("\nReport Summary:") - print(f"Total Customers: {report['statistics']['total_customers']}") - print(f"Average Age: {report['statistics']['average_age']}") - print(f"Average Spending: {report['statistics']['average_spending']}") - print(f"Average Loyalty Score: {report['statistics']['average_loyalty_score']}") - print("\nTop Customers by Loyalty Score:") - for i, customer in enumerate(report["top_customers"], 1): - print(f"{i}. {customer['name']} (Score: {customer['loyalty_score']})") - - print("\nWorkflow completed successfully!") - -finally: - # Always close the API client when done - api_client.close() diff --git a/tutorial/cosmotech-api/connection_setup.py b/tutorial/cosmotech-api/connection_setup.py index 7926cc6e..08956e76 100644 --- a/tutorial/cosmotech-api/connection_setup.py +++ b/tutorial/cosmotech-api/connection_setup.py @@ -1,64 +1,52 @@ # Example: Setting up connections to the CosmoTech API import os -from cosmotech.coal.cosmotech_api.connection import get_api_client +from cosmotech.coal.cosmotech_api.objects.connection import Connection from cosmotech.coal.utils.logger import LOGGER -# Method 1: Using API Key (set these environment variables before running) +# Method 1: API Key — requires CSM_API_URL and CSM_API_KEY os.environ["CSM_API_URL"] = "https://api.cosmotech.com" # Replace with your API URL os.environ["CSM_API_KEY"] = "your-api-key" # Replace with your actual API key -# Get the API client -api_client, connection_type = get_api_client() -LOGGER.info(f"Connected using: {connection_type}") +connection = Connection() +LOGGER.info(f"Connected using: {connection.api_type}") -# Use the client with various API instances +# Use api_client directly with SDK classes if needed from cosmotech_api.api.organization_api import OrganizationApi -org_api = OrganizationApi(api_client) - -# List organizations +org_api = OrganizationApi(connection.api_client) organizations = org_api.find_all_organizations() for org in organizations: print(f"Organization: {org.name} (ID: {org.id})") -# Don't forget to close the client when done -api_client.close() +# CoAL API wrapper classes inherit Connection and handle auth automatically: +# from cosmotech.coal.cosmotech_api.apis import WorkspaceApi, RunnerApi, DatasetApi +# ws_api = WorkspaceApi() # auth resolved from environment +# runner_api = RunnerApi() +# dataset_api = DatasetApi() -# Method 2: Using Azure Entra (set these environment variables before running) +# Method 2: Azure Entra — requires CSM_API_URL, CSM_API_SCOPE, +# AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, AZURE_TENANT_ID """ -os.environ["CSM_API_URL"] = "https://api.cosmotech.com" # Replace with your API URL -os.environ["CSM_API_SCOPE"] = "api://your-app-id/.default" # Replace with your API scope -os.environ["AZURE_CLIENT_ID"] = "your-client-id" # Replace with your client ID -os.environ["AZURE_CLIENT_SECRET"] = "your-client-secret" # Replace with your client secret -os.environ["AZURE_TENANT_ID"] = "your-tenant-id" # Replace with your tenant ID - -# Get the API client -api_client, connection_type = get_api_client() -LOGGER.info(f"Connected using: {connection_type}") - -# Use the client with various API instances -# ... - -# Don't forget to close the client when done -api_client.close() +os.environ["CSM_API_URL"] = "https://api.cosmotech.com" +os.environ["CSM_API_SCOPE"] = "api://your-app-id/.default" +os.environ["AZURE_CLIENT_ID"] = "your-client-id" +os.environ["AZURE_CLIENT_SECRET"] = "your-client-secret" +os.environ["AZURE_TENANT_ID"] = "your-tenant-id" + +connection = Connection() +LOGGER.info(f"Connected using: {connection.api_type}") """ -# Method 3: Using Keycloak (set these environment variables before running) +# Method 3: Keycloak — requires CSM_API_URL, IDP_BASE_URL, IDP_TENANT_ID, +# IDP_CLIENT_ID, IDP_CLIENT_SECRET """ -os.environ["CSM_API_URL"] = "https://api.cosmotech.com" # Replace with your API URL -os.environ["IDP_BASE_URL"] = "https://keycloak.example.com/auth/" # Replace with your Keycloak URL -os.environ["IDP_TENANT_ID"] = "your-realm" # Replace with your realm -os.environ["IDP_CLIENT_ID"] = "your-client-id" # Replace with your client ID -os.environ["IDP_CLIENT_SECRET"] = "your-client-secret" # Replace with your client secret - -# Get the API client -api_client, connection_type = get_api_client() -LOGGER.info(f"Connected using: {connection_type}") - -# Use the client with various API instances -# ... - -# Don't forget to close the client when done -api_client.close() +os.environ["CSM_API_URL"] = "https://api.cosmotech.com" +os.environ["IDP_BASE_URL"] = "https://keycloak.example.com/auth/" +os.environ["IDP_TENANT_ID"] = "your-realm" +os.environ["IDP_CLIENT_ID"] = "your-client-id" +os.environ["IDP_CLIENT_SECRET"] = "your-client-secret" + +connection = Connection() +LOGGER.info(f"Connected using: {connection.api_type}") """ diff --git a/tutorial/cosmotech-api/runner_operations.py b/tutorial/cosmotech-api/runner_operations.py index 27f290c9..10ab9d21 100644 --- a/tutorial/cosmotech-api/runner_operations.py +++ b/tutorial/cosmotech-api/runner_operations.py @@ -2,108 +2,44 @@ import os import pathlib -from cosmotech.coal.cosmotech_api.connection import get_api_client -from cosmotech.coal.cosmotech_api.runner import ( - download_datasets, - download_runner_data, - get_runner_data, - get_runner_parameters, -) +from cosmotech.coal.cosmotech_api.apis import RunnerApi +from cosmotech.coal.utils.configuration import Configuration from cosmotech.coal.utils.logger import LOGGER -# Set up environment variables for authentication os.environ["CSM_API_URL"] = "https://api.cosmotech.com" # Replace with your API URL os.environ["CSM_API_KEY"] = "your-api-key" # Replace with your actual API key -# Organization, workspace, and runner IDs organization_id = "your-organization-id" # Replace with your organization ID workspace_id = "your-workspace-id" # Replace with your workspace ID runner_id = "your-runner-id" # Replace with your runner ID -# Get the API client -api_client, connection_type = get_api_client() -LOGGER.info(f"Connected using: {connection_type}") - -try: - # Example 1: Get runner data - runner_data = get_runner_data(organization_id, workspace_id, runner_id) - print(f"Runner name: {runner_data.name}") - print(f"Runner ID: {runner_data.id}") - print(f"Runner state: {runner_data.state}") - - # Example 2: Get runner parameters - parameters = get_runner_parameters(runner_data) - print("\nRunner parameters:") - for param in parameters: - print(f" - {param['parameterId']}: {param['value']} (type: {param['varType']})") - - # Example 3: Download runner data (parameters and datasets) - # Create directories for parameters and datasets - param_dir = pathlib.Path("./runner_parameters") - dataset_dir = pathlib.Path("./runner_datasets") - param_dir.mkdir(exist_ok=True, parents=True) - dataset_dir.mkdir(exist_ok=True, parents=True) - - # Download runner data - result = download_runner_data( - organization_id=organization_id, - workspace_id=workspace_id, - runner_id=runner_id, - parameter_folder=str(param_dir), - dataset_folder=str(dataset_dir), - read_files=True, # Read file contents - parallel=True, # Download datasets in parallel - write_json=True, # Write parameters as JSON - write_csv=True, # Write parameters as CSV - fetch_dataset=True, # Fetch datasets - ) - - print("\nDownloaded runner data:") - print(f" - Parameters saved to: {param_dir}") - print(f" - Datasets saved to: {dataset_dir}") - - # Example 4: Working with specific datasets - if result["datasets"]: - print("\nDatasets associated with the runner:") - for dataset_id, dataset_info in result["datasets"].items(): - print(f" - Dataset ID: {dataset_id}") - print(f" Name: {dataset_info.get('name', 'N/A')}") - - # List files in the dataset - if "files" in dataset_info: - print(f" Files:") - for file_info in dataset_info["files"]: - print(f" - {file_info.get('name', 'N/A')}") - else: - print("\nNo datasets associated with this runner.") - - # Example 5: Download specific datasets - """ - from cosmotech.coal.cosmotech_api.runner import get_dataset_ids_from_runner - - # Get dataset IDs from the runner - dataset_ids = get_dataset_ids_from_runner(runner_data) - - if dataset_ids: - # Create a directory for the datasets - specific_dataset_dir = pathlib.Path("./specific_datasets") - specific_dataset_dir.mkdir(exist_ok=True, parents=True) - - # Download the datasets - datasets = download_datasets( - organization_id=organization_id, - workspace_id=workspace_id, - dataset_ids=dataset_ids, - read_files=True, - parallel=True, - ) - - print("\nDownloaded specific datasets:") - for dataset_id, dataset_info in datasets.items(): - print(f" - Dataset ID: {dataset_id}") - print(f" Name: {dataset_info.get('name', 'N/A')}") - """ - -finally: - # Always close the API client when done - api_client.close() +# Directories for downloaded data +param_dir = pathlib.Path("./runner_parameters") +dataset_dir = pathlib.Path("./runner_datasets") +param_dir.mkdir(exist_ok=True, parents=True) +dataset_dir.mkdir(exist_ok=True, parents=True) + +# Build a Configuration scoped to this runner +config = Configuration() +config.cosmotech.organization_id = organization_id +config.cosmotech.workspace_id = workspace_id +config.cosmotech.runner_id = runner_id +config.cosmotech.parameters_absolute_path = str(param_dir) +config.cosmotech.dataset_absolute_path = str(dataset_dir) + +runner_api = RunnerApi(config) + +# Example 1: Get runner metadata +metadata = runner_api.get_runner_metadata(runner_id=runner_id) +print(f"Runner name: {metadata.get('name')}") +print(f"Runner state: {metadata.get('state')}") + +# Optionally scope the returned fields: +# metadata = runner_api.get_runner_metadata( +# runner_id=runner_id, include=["parametersValues", "datasetList"] +# ) + +# Example 2: Download runner parameters and datasets +runner_api.download_runner_data(download_datasets=True) +print(f"Parameters saved to: {param_dir}") +print(f"Datasets saved to: {dataset_dir}") diff --git a/tutorial/cosmotech-api/twin_data_layer.py b/tutorial/cosmotech-api/twin_data_layer.py index 0036f658..b6635f0c 100644 --- a/tutorial/cosmotech-api/twin_data_layer.py +++ b/tutorial/cosmotech-api/twin_data_layer.py @@ -1,142 +1,72 @@ -# Example: Working with the Twin Data Layer in the CosmoTech API +# Example: Working with the Twin Data Layer via the CosmoTech API SDK +# +# NOTE: CoAL no longer provides Twin Data Layer helpers (CSVSourceFile, +# generate_query_insert, etc.). TDL operations must be performed directly +# through the cosmotech_api SDK's TwinGraphApi. import csv import os import pathlib from cosmotech_api.api.twin_graph_api import TwinGraphApi -from cosmotech.coal.cosmotech_api.connection import get_api_client -from cosmotech.coal.cosmotech_api.twin_data_layer import CSVSourceFile +from cosmotech.coal.cosmotech_api.objects.connection import Connection from cosmotech.coal.utils.logger import LOGGER -# Set up environment variables for authentication os.environ["CSM_API_URL"] = "https://api.cosmotech.com" # Replace with your API URL os.environ["CSM_API_KEY"] = "your-api-key" # Replace with your actual API key -# Organization and workspace IDs -organization_id = "your-organization-id" # Replace with your organization ID -workspace_id = "your-workspace-id" # Replace with your workspace ID -twin_graph_id = "your-twin-graph-id" # Replace with your twin graph ID - -# Get the API client -api_client, connection_type = get_api_client() -LOGGER.info(f"Connected using: {connection_type}") - -try: - # Create a TwinGraphApi instance - twin_graph_api = TwinGraphApi(api_client) - - # Example 1: Create sample CSV files for nodes and relationships - - # Create a directory for our sample data - data_dir = pathlib.Path("./tdl_sample_data") - data_dir.mkdir(exist_ok=True, parents=True) - - # Create a sample nodes CSV file (Person nodes) - persons_file = data_dir / "Person.csv" - with open(persons_file, "w", newline="") as f: - writer = csv.writer(f) - writer.writerow(["id", "name", "age", "city"]) - writer.writerow(["p1", "Alice", "30", "New York"]) - writer.writerow(["p2", "Bob", "25", "San Francisco"]) - writer.writerow(["p3", "Charlie", "35", "Chicago"]) - - # Create a sample relationships CSV file (KNOWS relationships) - knows_file = data_dir / "KNOWS.csv" - with open(knows_file, "w", newline="") as f: - writer = csv.writer(f) - writer.writerow(["src", "dest", "since"]) - writer.writerow(["p1", "p2", "2020"]) - writer.writerow(["p2", "p3", "2021"]) - writer.writerow(["p3", "p1", "2019"]) - - print(f"Created sample CSV files in {data_dir}") - - # Example 2: Parse CSV files and generate Cypher queries - - # Parse the nodes CSV file - person_csv = CSVSourceFile(persons_file) - print(f"Parsed {person_csv.object_type} CSV file:") - print(f" Is node: {person_csv.is_node}") - print(f" Fields: {person_csv.fields}") - print(f" ID column: {person_csv.id_column}") - - # Generate a Cypher query for creating nodes - person_query = person_csv.generate_query_insert() - print(f"\nGenerated Cypher query for {person_csv.object_type}:") - print(person_query) - - # Parse the relationships CSV file - knows_csv = CSVSourceFile(knows_file) - print(f"\nParsed {knows_csv.object_type} CSV file:") - print(f" Is node: {knows_csv.is_node}") - print(f" Fields: {knows_csv.fields}") - print(f" Source column: {knows_csv.source_column}") - print(f" Target column: {knows_csv.target_column}") - - # Generate a Cypher query for creating relationships - knows_query = knows_csv.generate_query_insert() - print(f"\nGenerated Cypher query for {knows_csv.object_type}:") - print(knows_query) - - # Example 3: Send data to the Twin Data Layer (commented out as it requires an actual twin graph) - """ - # For nodes, you would typically: - with open(persons_file, "r") as f: - reader = csv.DictReader(f) - for row in reader: - # Create parameters for the Cypher query - params = {k: v for k, v in row.items()} - - # Execute the query - twin_graph_api.run_twin_graph_cypher_query( - organization_id=organization_id, - workspace_id=workspace_id, - twin_graph_id=twin_graph_id, - twin_graph_cypher_query={ - "query": person_query, - "parameters": params - } - ) - - # For relationships, you would typically: - with open(knows_file, "r") as f: - reader = csv.DictReader(f) - for row in reader: - # Create parameters for the Cypher query - params = {k: v for k, v in row.items()} - - # Execute the query - twin_graph_api.run_twin_graph_cypher_query( - organization_id=organization_id, - workspace_id=workspace_id, - twin_graph_id=twin_graph_id, - twin_graph_cypher_query={ - "query": knows_query, - "parameters": params - } - ) - """ - - # Example 4: Query data from the Twin Data Layer (commented out as it requires an actual twin graph) - """ - # Execute a Cypher query to get all Person nodes - result = twin_graph_api.run_twin_graph_cypher_query( - organization_id=organization_id, - workspace_id=workspace_id, - twin_graph_id=twin_graph_id, - twin_graph_cypher_query={ - "query": "MATCH (p:Person) RETURN p.id, p.name, p.age, p.city", - "parameters": {} - } - ) - - # Process the results - print("\nPerson nodes in the Twin Data Layer:") - for record in result.records: - print(f" - {record}") - """ - -finally: - # Always close the API client when done - api_client.close() +organization_id = "your-organization-id" +workspace_id = "your-workspace-id" +twin_graph_id = "your-twin-graph-id" + +connection = Connection() +twin_graph_api = TwinGraphApi(connection.api_client) + +# Create sample CSV data for nodes and relationships +data_dir = pathlib.Path("./tdl_sample_data") +data_dir.mkdir(exist_ok=True, parents=True) + +persons_file = data_dir / "Person.csv" +with open(persons_file, "w", newline="") as f: + writer = csv.writer(f) + writer.writerow(["id", "name", "age", "city"]) + writer.writerow(["p1", "Alice", "30", "New York"]) + writer.writerow(["p2", "Bob", "25", "San Francisco"]) + +knows_file = data_dir / "KNOWS.csv" +with open(knows_file, "w", newline="") as f: + writer = csv.writer(f) + writer.writerow(["src", "dest", "since"]) + writer.writerow(["p1", "p2", "2020"]) + +# Example: send node rows to TDL using raw Cypher queries +create_person_query = ( + "MERGE (p:Person {id: $id}) " + "SET p.name = $name, p.age = $age, p.city = $city" +) + +""" +# Uncomment to run against an actual twin graph: +with open(persons_file, "r") as f: + for row in csv.DictReader(f): + twin_graph_api.run_twin_graph_cypher_query( + organization_id=organization_id, + workspace_id=workspace_id, + twin_graph_id=twin_graph_id, + twin_graph_cypher_query={"query": create_person_query, "parameters": row}, + ) + +create_knows_query = ( + "MATCH (a:Person {id: $src}), (b:Person {id: $dest}) " + "MERGE (a)-[r:KNOWS]->(b) SET r.since = $since" +) + +with open(knows_file, "r") as f: + for row in csv.DictReader(f): + twin_graph_api.run_twin_graph_cypher_query( + organization_id=organization_id, + workspace_id=workspace_id, + twin_graph_id=twin_graph_id, + twin_graph_cypher_query={"query": create_knows_query, "parameters": row}, + ) +""" diff --git a/tutorial/cosmotech-api/workspace_operations.py b/tutorial/cosmotech-api/workspace_operations.py index 90114ee1..e114185b 100644 --- a/tutorial/cosmotech-api/workspace_operations.py +++ b/tutorial/cosmotech-api/workspace_operations.py @@ -2,67 +2,52 @@ import os import pathlib -from cosmotech.coal.cosmotech_api.connection import get_api_client -from cosmotech.coal.cosmotech_api.workspace import ( - download_workspace_file, - list_workspace_files, - upload_workspace_file, -) +from cosmotech.coal.cosmotech_api.apis import WorkspaceApi from cosmotech.coal.utils.logger import LOGGER -# Set up environment variables for authentication os.environ["CSM_API_URL"] = "https://api.cosmotech.com" # Replace with your API URL os.environ["CSM_API_KEY"] = "your-api-key" # Replace with your actual API key -# Organization and workspace IDs organization_id = "your-organization-id" # Replace with your organization ID workspace_id = "your-workspace-id" # Replace with your workspace ID -# Get the API client -api_client, connection_type = get_api_client() -LOGGER.info(f"Connected using: {connection_type}") +ws_api = WorkspaceApi() +# Example 1: List workspace files with a given prefix +file_prefix = "data/" try: - # Example 1: List files in a workspace with a specific prefix - file_prefix = "data/" # List files in the "data" directory - try: - files = list_workspace_files(api_client, organization_id, workspace_id, file_prefix) - print(f"Files in workspace with prefix '{file_prefix}':") - for file in files: - print(f" - {file}") - except ValueError as e: - print(f"Error listing files: {e}") + files = ws_api.list_filtered_workspace_files(organization_id, workspace_id, file_prefix) + print(f"Files in workspace with prefix '{file_prefix}':") + for file in files: + print(f" - {file}") +except ValueError as e: + print(f"No files found: {e}") + +# Example 2: Download a file from the workspace +file_to_download = "data/sample.csv" # Replace with an actual file in your workspace +target_directory = pathlib.Path("./downloaded_files") +target_directory.mkdir(exist_ok=True, parents=True) - # Example 2: Download a file from the workspace - file_to_download = "data/sample.csv" # Replace with an actual file in your workspace - target_directory = pathlib.Path("./downloaded_files") - target_directory.mkdir(exist_ok=True, parents=True) - - try: - downloaded_file = download_workspace_file( - api_client, organization_id, workspace_id, file_to_download, target_directory - ) - print(f"Downloaded file to: {downloaded_file}") - except Exception as e: - print(f"Error downloading file: {e}") - - # Example 3: Upload a file to the workspace - file_to_upload = "./local_data/upload_sample.csv" # Replace with a local file path - workspace_destination = "data/uploaded/" # Destination in the workspace (ending with / to keep filename) +try: + local_path = ws_api.download_workspace_file( + organization_id, workspace_id, file_to_download, target_directory + ) + print(f"Downloaded file to: {local_path}") +except Exception as e: + print(f"Error downloading file: {e}") - try: - uploaded_file = upload_workspace_file( - api_client, - organization_id, - workspace_id, - file_to_upload, - workspace_destination, - overwrite=True, # Set to False to prevent overwriting existing files - ) - print(f"Uploaded file as: {uploaded_file}") - except Exception as e: - print(f"Error uploading file: {e}") +# Example 3: Upload a file to the workspace +file_to_upload = "./local_data/upload_sample.csv" # Replace with a local file path +workspace_destination = "data/uploaded/" # Trailing slash → original filename is kept -finally: - # Always close the API client when done - api_client.close() +try: + uploaded_name = ws_api.upload_workspace_file( + organization_id, + workspace_id, + file_to_upload, + workspace_destination, + overwrite=True, + ) + print(f"Uploaded file as: {uploaded_name}") +except Exception as e: + print(f"Error uploading file: {e}") diff --git a/tutorial/csm-data/api_env_variables.bash b/tutorial/csm-data/api_env_variables.bash index a39a1045..8ae97a22 100644 --- a/tutorial/csm-data/api_env_variables.bash +++ b/tutorial/csm-data/api_env_variables.bash @@ -1,7 +1,7 @@ # API connection export CSM_ORGANIZATION_ID="o-organization" export CSM_WORKSPACE_ID="w-workspace" -export CSM_SCENARIO_ID="s-scenario" +export CSM_RUNNER_ID="r-runner" # Paths export CSM_DATASET_ABSOLUTE_PATH="/path/to/dataset" diff --git a/tutorial/csm-data/complete_pipeline.bash b/tutorial/csm-data/complete_pipeline.bash index dc6fe7ec..577fe261 100644 --- a/tutorial/csm-data/complete_pipeline.bash +++ b/tutorial/csm-data/complete_pipeline.bash @@ -4,9 +4,7 @@ csm-data api run-load-data \ --workspace-id "$CSM_WORKSPACE_ID" \ --runner-id "$CSM_RUNNER_ID" \ --dataset-absolute-path "$CSM_DATASET_ABSOLUTE_PATH" \ - --parameters-absolute-path "$CSM_PARAMETERS_ABSOLUTE_PATH" \ - --write-json \ - --fetch-dataset + --parameters-absolute-path "$CSM_PARAMETERS_ABSOLUTE_PATH" # 2. Load data into the datastore for processing csm-data store load-csv-folder \ diff --git a/tutorial/csm-data/csm_orc_integration.json b/tutorial/csm-data/csm_orc_integration.json index 5b6133cb..328a826c 100644 --- a/tutorial/csm-data/csm_orc_integration.json +++ b/tutorial/csm-data/csm_orc_integration.json @@ -1,12 +1,10 @@ { "steps": [ { - "id": "download-scenario-data", + "id": "download-runner-data", "command": "csm-data", "arguments": [ - "api", "scenariorun-load-data", - "--write-json", - "--fetch-dataset" + "api", "run-load-data" ], "useSystemEnvironment": true }, @@ -14,13 +12,13 @@ "id": "run-simulation", "command": "python", "arguments": ["run_simulation.py"], - "precedents": ["download-scenario-data"] + "precedents": ["download-runner-data"] }, { "id": "send-results-to-adx", "command": "csm-data", "arguments": [ - "adx-send-scenariodata", + "adx-send-runnerdata", "--send-datasets", "--wait" ], diff --git a/tutorial/csm-data/run_load_data.bash b/tutorial/csm-data/run_load_data.bash index f5992a17..03897d00 100644 --- a/tutorial/csm-data/run_load_data.bash +++ b/tutorial/csm-data/run_load_data.bash @@ -3,7 +3,4 @@ csm-data api run-load-data \ --workspace-id "w-workspace" \ --runner-id "r-runner" \ --dataset-absolute-path "/path/to/dataset/folder" \ - --parameters-absolute-path "/path/to/parameters/folder" \ - --write-json \ - --write-csv \ - --fetch-dataset + --parameters-absolute-path "/path/to/parameters/folder" diff --git a/tutorial/datastore/parquet_files.py b/tutorial/datastore/parquet_files.py new file mode 100644 index 00000000..fa09d905 --- /dev/null +++ b/tutorial/datastore/parquet_files.py @@ -0,0 +1,51 @@ +import pathlib + +import pyarrow.parquet as pq + +from cosmotech.coal.store.parquet import convert_store_table_to_parquet, store_parquet_file +from cosmotech.coal.store.store import Store + +# Initialize the store +store = Store(reset=True) + +# --- Loading a Parquet file into the store --- + +# Create a sample parquet file for demonstration +sample_dir = pathlib.Path("./parquet_example") +sample_dir.mkdir(exist_ok=True, parents=True) +sample_file = sample_dir / "sales.parquet" + +import pyarrow as pa + +table = pa.table( + { + "region": ["North", "South", "East", "West"], + "product": ["Widget", "Gadget", "Widget", "Gadget"], + "units": [120, 85, 200, 60], + "revenue": [2400.0, 1275.0, 4000.0, 900.0], + } +) +pq.write_table(table, sample_file) + +# Load the parquet file into the store under the table name "sales" +store_parquet_file("sales", sample_file, store=store) + +# Query the loaded data +result = store.execute_query( + """ + SELECT product, SUM(units) AS total_units, SUM(revenue) AS total_revenue + FROM sales + GROUP BY product + ORDER BY total_revenue DESC +""" +) +print(result) + +# --- Exporting a store table back to Parquet --- + +output_dir = pathlib.Path("./parquet_output") +output_dir.mkdir(exist_ok=True, parents=True) + +# Write the "sales" table from the store to a parquet file +convert_store_table_to_parquet("sales", output_dir / "sales.parquet", store=store) +print(f"Exported store table 'sales' to {output_dir / 'sales.parquet'}") From f5902e07b9819f45e9c09c443956a1ab24f2cb0e Mon Sep 17 00:00:00 2001 From: Hector Jacinto Murillo Date: Mon, 10 Aug 2026 17:01:18 +0200 Subject: [PATCH 3/6] doc: apply minor corrections after updates - restablish ws_api calls details - update dataset_api calls - add new line (x2) to fix lists rendering - remove unused logger imports - remove deprecated TDL script - rename upload_workspace_file call args --- docs/tutorials/cosmotech-api.md | 47 +++++++++--- docs/tutorials/csm-data.md | 2 + tutorial/cosmotech-api/complete_workflow.py | 7 +- tutorial/cosmotech-api/runner_operations.py | 1 - tutorial/cosmotech-api/twin_data_layer.py | 72 ------------------- .../cosmotech-api/workspace_operations.py | 9 ++- 6 files changed, 48 insertions(+), 90 deletions(-) delete mode 100644 tutorial/cosmotech-api/twin_data_layer.py diff --git a/docs/tutorials/cosmotech-api.md b/docs/tutorials/cosmotech-api.md index 71d1257e..276ad5d9 100644 --- a/docs/tutorials/cosmotech-api.md +++ b/docs/tutorials/cosmotech-api.md @@ -108,15 +108,47 @@ Workspaces in the CosmoTech platform provide a way to organize and share files. ### Listing Files -`list_filtered_workspace_files` returns all workspace files whose `file_name` starts with the given prefix. It raises `ValueError` when no matching files are found. +`list_filtered_workspace_files` returns all workspace files whose `file_name` starts with the given prefix. It raises `ValueError` when no matching files are found: + +```python +files = ws_api.list_filtered_workspace_files( + organization_id, + workspace_id, + file_prefix +) +``` + +This is useful for finding files in a specific directory or with a specific naming pattern. ### Downloading Files -`download_workspace_file` writes the file content to `target_dir / file_name`, creating any necessary intermediate directories. +`download_workspace_file` writes the file content to `target_dir / file_name`, creating any necessary intermediate directories: + + +```python +local_path = ws_api.download_workspace_file( + organization_id, + workspace_id, + file_to_download, + target_directory +) +``` ### Uploading Files -`upload_workspace_file` uploads a single local file. The `workspace_path` parameter can be: +`upload_workspace_file` uploads a single local file: + +```python +uploaded_name = ws_api.upload_workspace_file( + organization_id, + workspace_id, + file_path, + workspace_path, + overwrite=True, +) +``` + +The `workspace_path` parameter can be: - A specific file path in the workspace - A directory path ending with `/`, in which case the original filename is preserved @@ -139,16 +171,15 @@ dataset_api = DatasetApi() # Upload a single file as a dataset dataset_api.upload_dataset( - organization_id=organization_id, - dataset_id=dataset_id, - file_path="/tmp/data/customers.csv", + dataset_name="customers", + as_files=["/tmp/data/customers.csv"], ) +dataset_id = dataset.id # Upload multiple parts from a folder (one part per file) dataset_api.upload_dataset_parts( - organization_id=organization_id, dataset_id=dataset_id, - folder_path="/tmp/data/parts/", + as_files=["/tmp/data/customers_p1.csv", "/tmp/data/customers_p2.csv"], ) # Download a dataset to a local directory diff --git a/docs/tutorials/csm-data.md b/docs/tutorials/csm-data.md index 2c726706..63ae65c6 100644 --- a/docs/tutorials/csm-data.md +++ b/docs/tutorials/csm-data.md @@ -110,6 +110,7 @@ The `adx-send-runnerdata` command enables sending runner data to Azure Data Expl ``` This command: + - Creates tables in ADX based on CSV files in the dataset and/or parameters folders - Ingests the data into those tables - Adds a `run` column with the runner ID for tracking @@ -131,6 +132,7 @@ The `store` command group provides tools for working with the CoAL datastore: ``` These commands allow you to: + - Load data from CSV files into the datastore - Load data from Parquet files into the datastore - Dump datastore contents to various destinations (S3, Azure, PostgreSQL) diff --git a/tutorial/cosmotech-api/complete_workflow.py b/tutorial/cosmotech-api/complete_workflow.py index 2cb7e72e..ea3c7fa3 100644 --- a/tutorial/cosmotech-api/complete_workflow.py +++ b/tutorial/cosmotech-api/complete_workflow.py @@ -6,7 +6,6 @@ from cosmotech.coal.cosmotech_api.apis import DatasetApi, RunnerApi, WorkspaceApi from cosmotech.coal.utils.configuration import Configuration -from cosmotech.coal.utils.logger import LOGGER os.environ["CSM_API_URL"] = "https://api.cosmotech.com" # Replace with your API URL os.environ["CSM_API_KEY"] = "your-api-key" # Replace with your actual API key @@ -88,10 +87,10 @@ try: dataset_api = DatasetApi(config) dataset_api.upload_dataset( - organization_id=organization_id, - dataset_id=output_dataset_id, - file_path=str(processed_file), + dataset_name="customers", + as_files=str(processed_file), ) + output_dataset_id = dataset_api.id print(f"Dataset {output_dataset_id} updated") except Exception as e: print(f"Error updating dataset: {e}") diff --git a/tutorial/cosmotech-api/runner_operations.py b/tutorial/cosmotech-api/runner_operations.py index 10ab9d21..5dc3e60c 100644 --- a/tutorial/cosmotech-api/runner_operations.py +++ b/tutorial/cosmotech-api/runner_operations.py @@ -4,7 +4,6 @@ from cosmotech.coal.cosmotech_api.apis import RunnerApi from cosmotech.coal.utils.configuration import Configuration -from cosmotech.coal.utils.logger import LOGGER os.environ["CSM_API_URL"] = "https://api.cosmotech.com" # Replace with your API URL os.environ["CSM_API_KEY"] = "your-api-key" # Replace with your actual API key diff --git a/tutorial/cosmotech-api/twin_data_layer.py b/tutorial/cosmotech-api/twin_data_layer.py deleted file mode 100644 index b6635f0c..00000000 --- a/tutorial/cosmotech-api/twin_data_layer.py +++ /dev/null @@ -1,72 +0,0 @@ -# Example: Working with the Twin Data Layer via the CosmoTech API SDK -# -# NOTE: CoAL no longer provides Twin Data Layer helpers (CSVSourceFile, -# generate_query_insert, etc.). TDL operations must be performed directly -# through the cosmotech_api SDK's TwinGraphApi. -import csv -import os -import pathlib - -from cosmotech_api.api.twin_graph_api import TwinGraphApi - -from cosmotech.coal.cosmotech_api.objects.connection import Connection -from cosmotech.coal.utils.logger import LOGGER - -os.environ["CSM_API_URL"] = "https://api.cosmotech.com" # Replace with your API URL -os.environ["CSM_API_KEY"] = "your-api-key" # Replace with your actual API key - -organization_id = "your-organization-id" -workspace_id = "your-workspace-id" -twin_graph_id = "your-twin-graph-id" - -connection = Connection() -twin_graph_api = TwinGraphApi(connection.api_client) - -# Create sample CSV data for nodes and relationships -data_dir = pathlib.Path("./tdl_sample_data") -data_dir.mkdir(exist_ok=True, parents=True) - -persons_file = data_dir / "Person.csv" -with open(persons_file, "w", newline="") as f: - writer = csv.writer(f) - writer.writerow(["id", "name", "age", "city"]) - writer.writerow(["p1", "Alice", "30", "New York"]) - writer.writerow(["p2", "Bob", "25", "San Francisco"]) - -knows_file = data_dir / "KNOWS.csv" -with open(knows_file, "w", newline="") as f: - writer = csv.writer(f) - writer.writerow(["src", "dest", "since"]) - writer.writerow(["p1", "p2", "2020"]) - -# Example: send node rows to TDL using raw Cypher queries -create_person_query = ( - "MERGE (p:Person {id: $id}) " - "SET p.name = $name, p.age = $age, p.city = $city" -) - -""" -# Uncomment to run against an actual twin graph: -with open(persons_file, "r") as f: - for row in csv.DictReader(f): - twin_graph_api.run_twin_graph_cypher_query( - organization_id=organization_id, - workspace_id=workspace_id, - twin_graph_id=twin_graph_id, - twin_graph_cypher_query={"query": create_person_query, "parameters": row}, - ) - -create_knows_query = ( - "MATCH (a:Person {id: $src}), (b:Person {id: $dest}) " - "MERGE (a)-[r:KNOWS]->(b) SET r.since = $since" -) - -with open(knows_file, "r") as f: - for row in csv.DictReader(f): - twin_graph_api.run_twin_graph_cypher_query( - organization_id=organization_id, - workspace_id=workspace_id, - twin_graph_id=twin_graph_id, - twin_graph_cypher_query={"query": create_knows_query, "parameters": row}, - ) -""" diff --git a/tutorial/cosmotech-api/workspace_operations.py b/tutorial/cosmotech-api/workspace_operations.py index e114185b..cc2a9695 100644 --- a/tutorial/cosmotech-api/workspace_operations.py +++ b/tutorial/cosmotech-api/workspace_operations.py @@ -3,7 +3,6 @@ import pathlib from cosmotech.coal.cosmotech_api.apis import WorkspaceApi -from cosmotech.coal.utils.logger import LOGGER os.environ["CSM_API_URL"] = "https://api.cosmotech.com" # Replace with your API URL os.environ["CSM_API_KEY"] = "your-api-key" # Replace with your actual API key @@ -37,15 +36,15 @@ print(f"Error downloading file: {e}") # Example 3: Upload a file to the workspace -file_to_upload = "./local_data/upload_sample.csv" # Replace with a local file path -workspace_destination = "data/uploaded/" # Trailing slash → original filename is kept +file_path = "./local_data/upload_sample.csv" # Replace with a local file path +workspace_path = "data/uploaded/" # Trailing slash → original filename is kept try: uploaded_name = ws_api.upload_workspace_file( organization_id, workspace_id, - file_to_upload, - workspace_destination, + file_path, + workspace_path, overwrite=True, ) print(f"Uploaded file as: {uploaded_name}") From 7a2f6b1622a9925a7bc26542eb27a59a168b62d4 Mon Sep 17 00:00:00 2001 From: Hector Jacinto Murillo Date: Tue, 11 Aug 2026 14:43:06 +0200 Subject: [PATCH 4/6] doc: apply black formatting to tutorial scripts --- tutorial/cosmotech-api/complete_workflow.py | 5 +---- tutorial/cosmotech-api/workspace_operations.py | 4 +--- 2 files changed, 2 insertions(+), 7 deletions(-) diff --git a/tutorial/cosmotech-api/complete_workflow.py b/tutorial/cosmotech-api/complete_workflow.py index ea3c7fa3..c87516e5 100644 --- a/tutorial/cosmotech-api/complete_workflow.py +++ b/tutorial/cosmotech-api/complete_workflow.py @@ -54,9 +54,7 @@ customers = [] with open(customers_file, "r") as f: for row in csv.DictReader(f): - row["loyalty_score"] = str( - round(int(row["spending"]) / 100 + (int(row["age"]) - 20) / 10, 1) - ) + row["loyalty_score"] = str(round(int(row["spending"]) / 100 + (int(row["age"]) - 20) / 10, 1)) customers.append(row) processed_file = processed_dir / "customers_with_loyalty.csv" @@ -115,4 +113,3 @@ print(f"Total customers: {report['statistics']['total_customers']}") print(f"Avg loyalty score: {report['statistics']['average_loyalty_score']}") print("\nWorkflow completed successfully!") - diff --git a/tutorial/cosmotech-api/workspace_operations.py b/tutorial/cosmotech-api/workspace_operations.py index cc2a9695..a851a43e 100644 --- a/tutorial/cosmotech-api/workspace_operations.py +++ b/tutorial/cosmotech-api/workspace_operations.py @@ -28,9 +28,7 @@ target_directory.mkdir(exist_ok=True, parents=True) try: - local_path = ws_api.download_workspace_file( - organization_id, workspace_id, file_to_download, target_directory - ) + local_path = ws_api.download_workspace_file(organization_id, workspace_id, file_to_download, target_directory) print(f"Downloaded file to: {local_path}") except Exception as e: print(f"Error downloading file: {e}") From 953cac5045b4b229171ad44be169b07a8d559769 Mon Sep 17 00:00:00 2001 From: Hector Jacinto Murillo Date: Thu, 13 Aug 2026 16:11:58 +0200 Subject: [PATCH 5/6] doc: document the Configuration system --- docs/tutorials/cosmotech-api.md | 32 ++++++++++ tutorial/cosmotech-api/coal-config.toml | 78 +++++++++++++++++++++++++ 2 files changed, 110 insertions(+) create mode 100644 tutorial/cosmotech-api/coal-config.toml diff --git a/docs/tutorials/cosmotech-api.md b/docs/tutorials/cosmotech-api.md index 276ad5d9..b433f3fd 100644 --- a/docs/tutorials/cosmotech-api.md +++ b/docs/tutorials/cosmotech-api.md @@ -98,6 +98,38 @@ Keycloak authentication requires these environment variables: !!! warning "API Client Lifecycle" Always close the API client when you're done using it to release resources. The best practice is to use a `try`/`finally` block to ensure the client is closed even if an error occurs. +## Configuration + +The CoAL configuration system is based on a centralized data dictionary used to manage platform settings and behaviors dynamically. It allows scripts to run without requiring users to manually define connection or output specifics every single time. Data is primarily sourced from a TOML file loaded into a Kubernetes ConfigMap. + +### Core mechanics + +- The Configuration singleton: CoAL provides a `ENVIRONMENT_CONFIGURATION` singleton that users can import this into their scripts (`from cosmotech.coal.utils.configuration import ENVIRONMENT_CONFIGURATION as EC`) to access properties using dot-notation, such as `EC.cosmotech.runner_id`. + +- Kubernetes (K8s) ConfigMap integration: To supply configuration inside a pod launched via a workflow, CoAL mounts a K8s ConfigMap containing the configuration file directly inside the container. + +- Automatic path loading: CoAL automatically attempts to load the TOML file at the specific path `/mnt/coal/coal-config.toml`, making K8s ConfigMap auto-mounts seamless. + +### Syntax + +The configuration uses the TOML format to support specific features: + +- **secrets**: Environment variables (e.g. credentials, `TWIN_CACHE_HOST`, or `IDP_BASE_URL`) that are loaded at startup. At import, they are initialized and then removed from the final configuration dictionary, so variables like `run_template_id` are accessed directly under `EC.cosmotech` rather than a "secrets" sub-dictionary. CosmoTech environment variables provided by the API are always loaded. + +- **env.**: Fetches environment variables dynamically at runtime (e.g. `env.POSTGRES_USER_PASSWORD`), unlike "secrets" which are resolved statically at import. + +- **Internal References ($)**: Allows configuration keys to reference other values in the same TOML file (e.g. `$postgres.host`). + +- **[[outputs]]**: Uses TOML double-bracket list syntax to define a series of output destinations (such as PostgreSQL, S3, or Azure Blob Storage) utilized by the ChannelSplitter to direct simulation results. + +- **Error handling**: CoAL handles internal configuration references (like `$config.path`) with proper error reporting such as the `ReferenceKeyError` exception for missing configuration references. + +### Configuration dictionary + +```toml title="Configuration TOML file" linenums="1" +--8<-- 'tutorial/cosmotech-api/coal-config.toml' +``` + ## Working with Workspaces Workspaces in the CosmoTech platform provide a way to organize and share files. `WorkspaceApi` offers methods for listing, downloading, and uploading files. diff --git a/tutorial/cosmotech-api/coal-config.toml b/tutorial/cosmotech-api/coal-config.toml new file mode 100644 index 00000000..9f5bc932 --- /dev/null +++ b/tutorial/cosmotech-api/coal-config.toml @@ -0,0 +1,78 @@ +# This an exemple of a coal-config.toml + +# the double [[ section indicate a list. This allows to define multiple outputs +# Each output define its type (currently supported: s3, az_storage, postgres) +# For each output, the configuration defines value needed to interact with a storage. Mandatory value are mark with a [M] + +[[outputs]] +type = "s3" # indicate the output channel to use +[outputs.conf.s3] +endpoint_url = # [M] url of the s3 server +access_key_id = # [M] the id used to connect (equivalent to a username) +secret_access_key = # [M] the key to connect (equivalent to a password) +bucket_name = # [M] name of the bucket. s3 bucket is a concept that define a ressource that holds data. +bucket_prefix = # This is a prefix that will be add in front of each upload file + # (this can allow to store in subfolder by adding a prefix ending with "/") +outputs_type = # indicate the type of the data push. Can be .parquet or .csv (default: .csv) +use_ssl = # indicate the use of ssl (default: True) +ssl_cert_bundle = # in case of a s3 using custom SSL certification; here is where to put the path to the custom pem bundle. + # (Can also be set to False to not verify SSL certificat) + +[[outputs]] +type = "az_storage" # indicate the output channel to use +[outputs.conf.azure] +account_name = # use to build azure storage URL +tenant_id = # Azure tenant ID +client_id = # Azure client ID +client_secret = # Azure secret +container_name = # Name of the container (equivalent to AWS Bucket name) +outputs_type = # indicate the type of the data push. Can be .parquet or .csv (default: .csv) +file_prefix = # This is a prefix that will be add in front of each upload file + # (this can allow to store in subfolder by adding a prefix ending with "/") + +[[outputs]] +type = "postgres" +[outputs.conf.postgres] +host = # Host URL of postgres server +port = # Port expose by postgres server +db_name = # Postgres db name +db_schema = # Postgres db schema +user_name = # postgres username +user_password = # posrgres password +table_prefix = # prefix used on table creation (useful in case of using a centralize DB to differentiate) +password_encoding = # boolean indicating if the password should be encoder (default: False) (used for password with special characters) + + +# The secrets section contains all values that will be replaced by environement variables +# The secrets section is transform at initialisation and then merge to root (the secrets sections no longer exist after transformation) + +# The cosmotech sub section is added by default (it's all the environment variables given by the API)" + +# # # DON'T ADD THIS IN THE FINAL FILE # # # + +[secrets.cosmotech] +dataset_absolute_path = "CSM_DATASET_ABSOLUTE_PATH" +parameters_absolute_path = "CSM_PARAMETERS_ABSOLUTE_PATH" +output_absolute_path = "CSM_OUTPUT_ABSOLUTE_PATH" +tmp_absolute_path = "CSM_TEMP_ABSOLUTE_PATH" +organization_id = "CSM_ORGANIZATION_ID" +workspace_id = "CSM_WORKSPACE_ID" +runner_id = "CSM_RUNNER_ID" +run_id = "CSM_RUN_ID" +run_template_id = "CSM_RUN_TEMPLATE_ID" + +[secrets.cosmotech.api] +url = "CSM_API_URL " +scope = "CSM_API_SCOPE" + +[secrets.cosmotech.twin_cache] +host = "TWIN_CACHE_HOST" +port = "TWIN_CACHE_PORT" +password = "TWIN_CACHE_PASSWORD" +username = "TWIN_CACHE_USERNAME" + +[secrets.cosmotech.idp] +base_url = "IDP_BASE_URL" +tenant_id = "IDP_TENANT_ID" +client_id = "IDP_CLIENT_ID" +client_secret = "IDP_CLIENT_SECRET" From a70ce7d1b0506b95ac3db6524b57e9da21258285 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Laurent=20Al=C3=A9p=C3=A9e?= Date: Thu, 13 Aug 2026 17:23:55 +0200 Subject: [PATCH 6/6] update code exemple to use coal configuration and new api objects --- README.md | 25 ++-- docs/tutorials/cosmotech-api.md | 60 +-------- docs/tutorials/datastore.md | 2 +- docs/tutorials/index.md | 2 +- tutorial/cosmotech-api/complete_workflow.py | 115 ------------------ tutorial/cosmotech-api/connection_setup.py | 52 -------- tutorial/cosmotech-api/dataset_operations.py | 24 ++++ tutorial/cosmotech-api/runner_operations.py | 41 ++----- .../cosmotech-api/workspace_operations.py | 28 +++-- 9 files changed, 66 insertions(+), 283 deletions(-) delete mode 100644 tutorial/cosmotech-api/complete_workflow.py delete mode 100644 tutorial/cosmotech-api/connection_setup.py create mode 100644 tutorial/cosmotech-api/dataset_operations.py diff --git a/README.md b/README.md index a998c5ae..d0132acd 100644 --- a/README.md +++ b/README.md @@ -52,36 +52,29 @@ CoAL provides comprehensive tools for interacting with the CosmoTech API, allowi - Authenticate with different identity providers (API Key, Azure Entra, Keycloak) - Manage workspaces and files -- Work with the Twin Data Layer for graph data - Handle runners and runs - Process and transform data - Build end-to-end workflows -```python -import os -from cosmotech.coal.cosmotech_api.connection import get_api_client - -# Set up environment variables for authentication -os.environ["CSM_API_URL"] = "https://api.cosmotech.com" # Replace with your API URL -os.environ["CSM_API_KEY"] = "your-api-key" # Replace with your actual API key - -# Get the API client -api_client, connection_type = get_api_client() -print(f"Connected using: {connection_type}") +CoAL use a Configuration that loads environement variables. This Allows to integrate authentication directly in api clients. Below the intantiation of OrganizationApi integrate the authentiation via the configuration file. +```python # Use the client with various API instances -from cosmotech_api.api.organization_api import OrganizationApi -org_api = OrganizationApi(api_client) +from cosmotech.coal.cosmotech_api.apis import OrganizationApi +org_api = OrganizationApi() # List organizations organizations = org_api.find_all_organizations() for org in organizations: print(f"Organization: {org.name} (ID: {org.id})") -# Don't forget to close the client when done -api_client.close() ``` +### Configuration + +CoAL uses a Configuration, it contains default configuration (all environment variable loaded for a Cosmotech RUN) and can be augmented by loading a .toml file. +The configuration is the tool that allows the seemless connection to the cosmotech API. + ### Other Components - **coal**: Core library with modules for API interaction, data management, etc. diff --git a/docs/tutorials/cosmotech-api.md b/docs/tutorials/cosmotech-api.md index b433f3fd..2937a228 100644 --- a/docs/tutorials/cosmotech-api.md +++ b/docs/tutorials/cosmotech-api.md @@ -50,11 +50,6 @@ The first step in working with the CosmoTech API is establishing a connection. C - Keycloak authentication The `Connection` class automatically detects which authentication method to use based on the environment variables present. - -```python title="Basic connection setup" linenums="1" ---8<-- 'tutorial/cosmotech-api/connection_setup.py' -``` - All API wrapper classes (`WorkspaceApi`, `RunnerApi`, `DatasetApi`, …) extend `Connection` and set themselves up automatically — you do not need to create the `Connection` separately unless you want direct access to the raw `ApiClient`. ```python @@ -66,7 +61,7 @@ dataset_api = DatasetApi() ``` !!! tip "Environment Variables" - You can set environment variables in your code for testing, but in production environments, it's better to set them at the system or container level for security. + You can set environment variables in your code for testing, but in production environments, it's better to set them at the container level using Coal configuration. Coal configuration uses a combination of Kubernetes ConfigMaps and Secrets to setup the environnement. ### API Key Authentication @@ -197,71 +192,26 @@ The `workspace_path` parameter can be: `DatasetApi` provides helpers for uploading datasets and managing their parts (files that compose the dataset). ```python title="Dataset upload" linenums="1" -from cosmotech.coal.cosmotech_api.apis import DatasetApi - -dataset_api = DatasetApi() - -# Upload a single file as a dataset -dataset_api.upload_dataset( - dataset_name="customers", - as_files=["/tmp/data/customers.csv"], -) -dataset_id = dataset.id - -# Upload multiple parts from a folder (one part per file) -dataset_api.upload_dataset_parts( - dataset_id=dataset_id, - as_files=["/tmp/data/customers_p1.csv", "/tmp/data/customers_p2.csv"], -) - -# Download a dataset to a local directory -dataset_api.download_dataset( - dataset_id=dataset_id, -) +--8<-- 'tutorial/cosmotech-api/dataset_operations.py' ``` !!! info "Dataset Parts" When uploading parts, the part name is derived from the filename without its extension. -## Runner and Run Management +## Runner Management -Runners and runs are central concepts in the CosmoTech platform. `RunnerApi` provides methods for retrieving runner metadata and downloading all associated data (parameters and datasets). +Runners are central concepts in the CosmoTech platform. `RunnerApi` provides methods for retrieving runner metadata and downloading all associated data (parameters and datasets). ```python title="Runner operations" linenums="1" --8<-- 'tutorial/cosmotech-api/runner_operations.py' ``` -## Complete Workflow Example - -Putting it all together, here's a typical end-to-end workflow for a CosmoTech data processing pipeline: - -```python title="Complete workflow" linenums="1" ---8<-- 'tutorial/cosmotech-api/complete_workflow.py' -``` - -This workflow: - -1. Downloads runner parameters and associated datasets -2. Processes the data (application-specific logic) -3. Uploads processed results to the workspace -4. Updates a dataset with the processed output parts - -!!! tip "Real-world Workflows" - In real-world scenarios, you might: - - - Use more complex data transformations - - Integrate with other Python code or services - - Implement error handling and retries - - Add logging and monitoring - - Parallelize operations for better performance - ## Best Practices and Tips ### Authentication -- Use environment variables for credentials - Implement proper secret management in production -- Always close API clients when done +- Use Coal configuration secrets loading for credentials ### Error Handling diff --git a/docs/tutorials/datastore.md b/docs/tutorials/datastore.md index f65cdadc..390d2b1c 100644 --- a/docs/tutorials/datastore.md +++ b/docs/tutorials/datastore.md @@ -132,7 +132,7 @@ The datastore provides specialized adapters for working with various data format !!! tip "Store initialization" - Use `reset=True` when you want to start with a fresh database - - Omit the reset parameter or set it to `False` when you want to maintain data between runs + - Omit the reset parameter or set it to `False` when you want to maintain data between steps - Specify a custom location with the `store_location` parameter if needed ```python title="Store initialization options" linenums="1" diff --git a/docs/tutorials/index.md b/docs/tutorials/index.md index d5bbd118..585a4b0f 100644 --- a/docs/tutorials/index.md +++ b/docs/tutorials/index.md @@ -43,7 +43,7 @@ The datastore is your friend to keep data between orchestration steps. It comes :material-api: __CosmoTech API__ --- -Learn how to interact with the CosmoTech API directly: authentication, workspaces, Twin Data Layer, and more. +Learn how to interact with the CosmoTech API directly: authentication, workspaces, runners and datasets. ---