Adding support for semantic layer apis

src/drs/cli.py

@@ -40,6 +40,7 @@
     reflection,
     role,
     schema,
+    semantic,
     setup,
     space,
     tag,
@@ -60,6 +61,7 @@
 app.add_typer(query.app, name="query")
 app.add_typer(folder.app, name="folder")
 app.add_typer(schema.app, name="schema")
+app.add_typer(semantic.app, name="semantic")
 app.add_typer(wiki.app, name="wiki")
 app.add_typer(tag.app, name="tag")
 app.add_typer(reflection.app, name="reflection")

src/drs/client.py

@@ -70,6 +70,10 @@ def _v3(self, path: str) -> str:
     def _v1(self, path: str) -> str:
         return f"{self.config.uri}/v1{path}"
 
+    def _v1_project(self, path: str) -> str:
+        """Project-scoped v1 URL: /v1/projects/{pid}/..."""
+        return f"{self.config.uri}/v1/projects/{self.config.project_id}{path}"
+
     # -- HTTP helpers with retry --
 
     async def _request_with_retry(self, method: str, url: str, **kwargs: Any) -> httpx.Response:
@@ -224,6 +228,10 @@ async def update_catalog_entity(self, entity_id: str, body: dict) -> dict:
         """Update a catalog entity. PUT /catalog/{id}."""
         return await self._put(self._v3(f"/catalog/{entity_id}"), json=body)
 
+    async def format_catalog_table(self, entity_id: str, body: dict) -> dict:
+        """Format a file or folder as a physical dataset. POST /catalog/{id}."""
+        return await self._post(self._v3(f"/catalog/{entity_id}"), json=body)
+
     async def delete_catalog_entity(self, entity_id: str, tag: str | None = None) -> dict:
         """Delete a catalog entity. DELETE /catalog/{id}."""
         params = {"tag": tag} if tag else None
@@ -252,6 +260,95 @@ async def set_tags(self, entity_id: str, tags: list[str], version: int | None =
             body["version"] = version
         return await self._post(self._v3(f"/catalog/{entity_id}/collaboration/tag"), json=body)
 
+    # -- Semantic layer (v4) --
+
+    async def delete_semantic_layer(self) -> dict:
+        return await self._delete(self._v1_project("/semantic-layer"))
+
+    async def initialize_semantic_layer(self, body: dict) -> dict:
+        return await self._post(self._v1_project("/semantic-layer/initialize"), json=body)
+
+    async def deploy_semantic_layer_entities(self, task_id: str) -> dict:
+        return await self._post(self._v1_project("/semantic-layer/deploy"), json={"taskId": task_id})
+
+    async def abandon_semantic_layer_entities(self, task_id: str) -> dict:
+        return await self._post(self._v1_project("/semantic-layer/abandon"), json={"taskId": task_id})
+
+    async def cancel_semantic_layer_task(self, task_id: str) -> dict:
+        return await self._post(self._v1_project("/semantic-layer/cancel"), json={"taskId": task_id})
+
+    async def list_semantic_layer_tasks(self, task_state: str | None = None) -> dict:
+        params = {"taskState": task_state} if task_state else None
+        return await self._get(self._v1_project("/semantic-layer/tasks"), params=params)
+
+    async def add_semantic_layer_task(self, body: dict) -> dict:
+        return await self._post(self._v1_project("/semantic-layer/tasks"), json=body)
+
+    async def get_semantic_layer_task(self, task_id: str) -> dict:
+        return await self._get(self._v1_project(f"/semantic-layer/tasks/{task_id}"))
+
+    async def list_semantic_layer_entities(
+        self,
+        entity_type: str,
+        task_id: str | None = None,
+        page_token: str | None = None,
+        limit: int | None = None,
+    ) -> dict:
+        params: dict[str, Any] = {"EntityType": entity_type}
+        if task_id:
+            params["taskId"] = task_id
+        if page_token:
+            params["pageToken"] = page_token
+        if limit is not None:
+            params["limit"] = limit
+        return await self._get(self._v1_project("/semantic-layer/entities"), params=params)
+
+    async def add_semantic_layer_entity(self, body: dict, task_id: str | None = None) -> dict:
+        url = self._v1_project("/semantic-layer/entities")
+        if task_id:
+            url = f"{url}?taskId={task_id}"
+        return await self._post(url, json=body)
+
+    async def get_semantic_layer_entity(
+        self,
+        entity_id: str,
+        entity_type: str,
+        task_id: str | None = None,
+        include_relationships: bool = False,
+    ) -> dict:
+        params: dict[str, Any] = {"EntityType": entity_type}
+        if task_id:
+            params["taskId"] = task_id
+        if include_relationships:
+            params["includeRelationships"] = "true"
+        return await self._get(self._v1_project(f"/semantic-layer/entities/{entity_id}"), params=params)
+
+    async def update_semantic_layer_entity(self, entity_id: str, body: dict, task_id: str | None = None) -> dict:
+        url = self._v1_project(f"/semantic-layer/entities/{entity_id}")
+        params = f"?taskId={task_id}" if task_id else ""
+        return await self._put(f"{url}{params}", json=body)
+
+    async def delete_semantic_layer_entity(self, entity_id: str, entity_type: str, task_id: str | None = None) -> dict:
+        params: dict[str, Any] = {"EntityType": entity_type}
+        if task_id:
+            params["taskId"] = task_id
+        return await self._delete(self._v1_project(f"/semantic-layer/entities/{entity_id}"), params=params)
+
+    async def get_semantic_layer_scope(self) -> dict:
+        return await self._get(self._v1_project("/semantic-layer/scope"))
+
+    async def patch_semantic_layer_scope(self, body: dict) -> dict:
+        logger.debug("PATCH %s body=%s", self._v1_project("/semantic-layer/scope"), body)
+        resp = await self._request_with_retry("PATCH", self._v1_project("/semantic-layer/scope"), json=body)
+        logger.debug(
+            "PATCH %s → %d (%d bytes)",
+            self._v1_project("/semantic-layer/scope"),
+            resp.status_code,
+            len(resp.content),
+        )
+        resp.raise_for_status()
+        return resp.json()
+
     # -- Reflections (v3) --
 
     async def get_reflection(self, reflection_id: str) -> dict:

src/drs/commands/folder.py

@@ -18,14 +18,22 @@
 from __future__ import annotations
 
 import asyncio
+from pathlib import Path
 
 import httpx
 import typer
 
 from drs.client import DremioClient
 from drs.commands.query import run_query
 from drs.output import OutputFormat, error, output, warn
-from drs.utils import DremioAPIError, NestedPathUnsupported, handle_api_error, parse_path, quote_path_sql
+from drs.utils import (
+    DremioAPIError,
+    NestedPathUnsupported,
+    handle_api_error,
+    parse_path,
+    quote_path_sql,
+    sanitize_input,
+)
 
 app = typer.Typer(
     help="Manage nested folders and list top-level catalog entities. Use `dremio space` for top-level spaces.",
@@ -114,6 +122,91 @@ async def grants(client: DremioClient, path: str) -> dict:
     }
 
 
+def _split_slash_path(path: str) -> list[str]:
+    """Split a slash-separated relative path into validated path parts."""
+    sanitize_input(path, "path")
+    parts = [part.strip() for part in path.split("/") if part.strip()]
+    if not parts:
+        raise ValueError("Path is empty.")
+    if any(part in {".", ".."} for part in parts):
+        raise ValueError(f"Invalid path '{path}': '.' and '..' path segments are not allowed.")
+    return parts
+
+
+async def promote_path_as_table(client: DremioClient, path_parts: list[str], format_type: str = "Delta") -> dict:
+    """Format a file or folder as a physical dataset using the Catalog API."""
+    try:
+        entity = await client.get_catalog_by_path(path_parts)
+    except httpx.HTTPStatusError as exc:
+        raise handle_api_error(exc) from exc
+
+    body = {
+        "entityType": "dataset",
+        "type": "PHYSICAL_DATASET",
+        # Use the caller-resolved catalog path parts directly. The lookup response
+        # can normalize or collapse file-source segments containing dots, which
+        # breaks the subsequent format request for file/folder sources.
+        "path": path_parts,
+        "format": {"type": format_type},
+    }
+
+    try:
+        return await client.format_catalog_table(entity["id"], body)
+    except httpx.HTTPStatusError as exc:
+        raise handle_api_error(exc) from exc
+
+
+async def promote_folder(client: DremioClient, path: str, format_type: str = "Delta") -> dict:
+    """Format a dot-separated file or folder path as a physical dataset."""
+    return await promote_path_as_table(client, parse_path(path), format_type=format_type)
+
+
+async def promote_from_file(
+    client: DremioClient,
+    paths_file: Path,
+    source: str,
+    under: str | None = None,
+    format_type: str = "Delta",
+) -> dict:
+    """Format multiple slash-separated relative paths from a file as datasets."""
+    source_part = sanitize_input(source.strip(), "source")
+    if not source_part:
+        raise ValueError("Source is empty.")
+
+    base_parts = [source_part]
+    if under:
+        base_parts.extend(_split_slash_path(under))
+
+    results: list[dict] = []
+    for line_no, raw in enumerate(paths_file.read_text(encoding="utf-8").splitlines(), start=1):
+        raw = raw.strip()
+        if not raw or raw.startswith("#"):
+            continue
+
+        rel_parts = _split_slash_path(raw)
+        full_parts = [*base_parts, *rel_parts]
+        result = await promote_path_as_table(client, full_parts, format_type=format_type)
+        results.append(
+            {
+                "line": line_no,
+                "input": raw,
+                "path": result.get("path", full_parts),
+                "id": result.get("id"),
+                "entityType": result.get("entityType"),
+                "type": result.get("type"),
+                "format": result.get("format"),
+            }
+        )
+
+    return {
+        "source": source_part,
+        "under": under,
+        "formatType": format_type,
+        "count": len(results),
+        "results": results,
+    }
+
+
 # -- CLI wrappers --
 
 
@@ -205,3 +298,55 @@ def cli_grants(
     """Show ACL grants on a catalog entity."""
     client = _get_client()
     _run_command(grants(client, path), client, fmt)
+
+
+@app.command("promote")
+def cli_promote(
+    path: str = typer.Argument(
+        None,
+        help="Dot-separated file or folder path to format as a table (e.g., source.folder.table_dir)",
+    ),
+    paths_file: Path | None = typer.Option(
+        None,
+        "--paths-file",
+        exists=True,
+        file_okay=True,
+        dir_okay=False,
+        readable=True,
+        resolve_path=True,
+        help="File containing slash-separated relative paths, one per line",
+    ),
+    source: str | None = typer.Option(
+        None,
+        "--source",
+        help="Source name to prefix to every line from --paths-file",
+    ),
+    under: str | None = typer.Option(
+        None,
+        "--under",
+        help="Optional slash-separated base path under the source for every line from --paths-file",
+    ),
+    format_type: str = typer.Option("Delta", "--format-type", help="Dataset format type to promote as"),
+    fmt: OutputFormat = typer.Option(OutputFormat.json, "--output", "-o", help="Output format"),
+    fields: str = typer.Option(None, "--fields", "-f", help="Comma-separated fields to include"),
+) -> None:
+    """Format a file or folder as a table using the Catalog API.
+
+    For batch promotion, pass --paths-file with slash-separated relative paths and
+    prefix them with --source and optional --under.
+    """
+    if bool(path) == bool(paths_file):
+        error("Provide exactly one of PATH or --paths-file.")
+        raise typer.Exit(1)
+
+    if paths_file and not source:
+        error("--source is required when using --paths-file.")
+        raise typer.Exit(1)
+
+    client = _get_client()
+    if paths_file:
+        _run_command(
+            promote_from_file(client, paths_file, source, under=under, format_type=format_type), client, fmt, fields
+        )
+        return
+    _run_command(promote_folder(client, path, format_type=format_type), client, fmt, fields=fields)

src/drs/commands/query.py

@@ -18,6 +18,7 @@
 from __future__ import annotations
 
 import asyncio
+import sys
 from pathlib import Path
 from typing import Annotated, Any
 
@@ -134,17 +135,8 @@ async def _execute():
 def cli_run(
     sql: str | None = typer.Argument(None, help="SQL query to execute (use '-' to read from stdin)"),
     file: Annotated[
-        Path | None,
-        typer.Option(
-            "--file",
-            exists=True,
-            file_okay=True,
-            dir_okay=False,
-            writable=False,
-            readable=True,
-            allow_dash=True,
-            help="Path to a SQL file to execute (use '-' for stdin)",
-        ),
+        typer.FileText|None,
+        typer.Option(help="Path to a SQL file to execute (use '-' for stdin)"),
     ] = None,
     context: str = typer.Option(None, help="Dot-separated default schema context (e.g., myspace.folder)"),
     fmt: OutputFormat = typer.Option(OutputFormat.json, "--output", "-o", help="Output format"),
@@ -171,9 +163,11 @@ def cli_run(
         if sql is not None:
             error("Cannot specify both a SQL argument and --file.")
             raise typer.Exit(1)
-        sql = file.read_text().strip()
+        sql = file.read().strip()
     elif sql is not None:
         sql = sql.strip()
+        if sql == "-":
+            sql = sys.stdin.read().strip()
 
     if not sql:
         error("SQL query is empty. Provide SQL as an argument, --file path, or pipe via stdin (use '-' for stdin).")