Added metadata search API

csst_fs/__init__.py

@@ -5,4 +5,5 @@ from .table import fsspec_table
 from . import s3_fs
 from . import fs
 from .commit import level2
-from .star_catalog import search
+from .star_catalog import search
\ No newline at end of file
+from .metadata import search
\ No newline at end of file

csst_fs/metadata/search.py

+import os
+import requests
+from typing import Dict, Any, List
+
+def query_metadata(
+    filter: Dict[str, Any],
+    key: List[str],
+) -> List[Dict[str, Any]]:
+    """
+    Query for file info by metadata values.
+
+    Args:
+        filter: The filter dict described below.
+        key: A list of string values, corresponding to metadata keys that should be included in the output.
+    Returns:
+        A List[Dict] of matching documents containing a file_path value and the keys set as 'key' parameter under 'metadata'.
+        E.g. with key = ["dataset", "instrument", "obs_group", "obs_id"]
+            then returns:
+            [
+                {
+                    "file_path": "CSST_L0/MSC/SCI/60310/10100000000/MS/CSST_MSC_MS_SCIE_20290225043953_20290225044223_10100000000_03_L0_V01.fits",
+                    "metadata": {
+                        "dataset":"csst-msc-c11-1000sqdeg-wide-test-v2",
+                        "instrument":"MSC",
+                        "obs_group":"W1",
+                        "obs_id":"10200000000"
+                    },
+                },
+            ]
+    """
+
+    if not filter:
+        raise ValueError("Filter cannot be empty")
+
+    api_url = os.getenv("METADATA_SEARCH_API_URL")
+    if not api_url:
+        raise RuntimeError("METADATA_SEARCH_API_URL environment variable is not set")
+
+    endpoint = f"{api_url}/datasync/metadata/query"
+
+    payload = {
+        "filter": filter,
+        "key": key,
+    }
+
+    try:
+        response = requests.post(endpoint, json=payload, timeout=30)
+        response.raise_for_status()
+    except requests.RequestException as e:
+        raise RuntimeError(f"Metadata query failed: {e}")
+
+    data = response.json()
+
+    if not data.get("success") or "result" not in data:
+        raise RuntimeError(f"Unexpected API response: {data}")
+
+    return data["result"]