---
title: "Catalog Code Examples"
slug: "catalog-code-examples"
description: "Explore runnable scripts for VAST 5.4.4.0 and 5.5.0.1, with setup guides and tips on query limits and directory listings for efficient catalog management."
updated: 2026-10-28T02:14:35Z
published: 2026-10-28T02:14:35Z
canonical: "docs.vastdata.com/catalog-code-examples"
---

> ## Documentation Index
> Fetch the complete documentation index at: https://docs.vastdata.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Catalog Code Examples

## Catalog KB examples

#### Runnable examples for the guide [Querying the VAST Catalog](/develop/docs/querying-the-vast-catalog)

Three runnable scripts, in easy-to-expert order. All were verified live against VAST 5.4.4.0 and VAST 5.5.0.1.

| Script | Path | Needs | Shows |
| --- | --- | --- | --- |
| `query_vms_catalog.py` | VMS REST API | VMS creds only | `bigcatalogconfig/query_data`: `path` anchor + `filters` (`eq`/`startswith`), cluster-wide view |
| `first_query.py` | VAST DB SDK | S3 key + `s3:Tabular*` policy | `tx.catalog()`, fast directory listing (equality + trailing slash), `--types`, `--schema` |
| `catalog_report.py` | VAST DB SDK | S3 key + `s3:Tabular*` policy | Pushdown recipes: largest files, owner, changed-since, extension, tagged (`user_tags_count`) |

Setup for the SDK scripts:

```plaintext
pip install vastdb pyarrow pandas ibis-framework
export VASTDB_ENDPOINT='http://<cluster-db-data-vip>'   # data VIP, not the VMS address
export VASTDB_ACCESS_KEY='<key>'
export VASTDB_SECRET_KEY='<secret>'
```

Setup for the VMS script:

```plaintext
pip install vastpy
export VMS_ADDRESS='<vms-host>'    # bare hostname
export VMS_USER='admin'
export VMS_PASSWORD='********'
```

Quick tour:

```plaintext
python query_vms_catalog.py --path / --type TABLE --top 20
python first_query.py --types
python first_query.py --path /projects/
python catalog_report.py --largest --under /projects/
python catalog_report.py --tagged
```

Remember the two speed rules (measured cost model in the guide's §3, Error 3): queries return early once `--limit` is satisfied but sweep the whole catalog when it isn't — keep `--limit` close to what you need; and directory listing is `parent_path` **equality with trailing slash** (without the slash you silently get zero rows).

#### catalog_report.py

```python
#!/usr/bin/env python3
"""
catalog_report.py: answer the common "what's on my cluster?" questions from the
VAST Catalog with the vastdb SDK.
All selectors AND-combine into one pushed-down predicate. Verified on VAST
5.4.4.0 (vastdb 2.0.14).
SETUP
    pip install vastdb pyarrow pandas ibis-framework
    export VASTDB_ENDPOINT='http://<cluster-db-data-vip>'   # data VIP, not the VMS address
    export VASTDB_ACCESS_KEY='<access-key>'
    export VASTDB_SECRET_KEY='<secret-key>'
USAGE
    python catalog_report.py --largest --under /projects/           # biggest files
    python catalog_report.py --owner root --under /home/            # one owner's files
    python catalog_report.py --changed-since '2026-08-01'           # recent changes
    python catalog_report.py --ext log --min-size 100M              # big .log files
    python catalog_report.py --tagged                               # elements with user tags
NOTES
    Every query is a capped catalog scan: it returns early once --limit rows
    have matched, and sweeps the whole catalog (~15s on a 3B-row cluster) when
    they haven't. Fetching very large result sets adds transfer time on top —
    that's the query that runs for minutes. So: keep --limit close to what you
    need, and anchor --under as deep as you can to keep match density high.
    See ../03-troubleshooting.md (Error 3) for the measured cost model.
"""
import argparse
import os
import re
import sys
from datetime import datetime
try:
    import ibis
    import vastdb
    from vastdb.config import BackoffConfig
    from ibis import _
except ImportError:
    sys.exit("Missing deps. Run: pip install vastdb pyarrow pandas ibis-framework")
COLUMNS = ["parent_path", "name", "element_type", "size", "used", "mtime",
           "owner_name", "uid", "extension"]
SIZE_UNITS = {"": 1, "K": 2**10, "M": 2**20, "G": 2**30, "T": 2**40}
def parse_size(text):
    m = re.fullmatch(r"(\d+)([KMGT]?)B?", text.strip().upper())
    if not m:
        sys.exit(f"can't parse size {text!r} (try 500M, 2G, ...)")
    return int(m.group(1)) * SIZE_UNITS[m.group(2)]
def parse_when(text):
    for fmt in ("%Y-%m-%d %H:%M", "%Y-%m-%d"):
        try:
            return datetime.strptime(text, fmt)
        except ValueError:
            continue
    sys.exit(f"can't parse time {text!r} (use 'YYYY-MM-DD' or 'YYYY-MM-DD HH:MM', UTC)")
def human(n):
    for unit in ("B", "KiB", "MiB", "GiB", "TiB"):
        if abs(n) < 1024 or unit == "TiB":
            return f"{n:.1f} {unit}" if unit != "B" else f"{int(n)} B"
        n /= 1024
def connect(timeout):
    endpoint = os.environ.get("VASTDB_ENDPOINT")
    access = os.environ.get("VASTDB_ACCESS_KEY")
    secret = os.environ.get("VASTDB_SECRET_KEY")
    if not (endpoint and access and secret):
        sys.exit("Set $VASTDB_ENDPOINT, $VASTDB_ACCESS_KEY, $VASTDB_SECRET_KEY.")
    return vastdb.connect(endpoint=endpoint, access=access, secret=secret,
                          timeout=timeout, backoff_config=BackoffConfig(max_tries=1))
def build_predicate(args):
    parts = []
    if args.under and args.under != "/":
        prefix = args.under if args.under.endswith("/") else args.under + "/"
        parts.append(_.parent_path.startswith(prefix))
    if args.owner:
        parts.append(_.owner_name == args.owner)
    if args.uid is not None:
        parts.append(_.uid == args.uid)
    if args.ext:
        parts.append(_.extension == args.ext.lstrip("."))
    if args.changed_since:
        parts.append(_.mtime >= ibis.literal(parse_when(args.changed_since),
                                             type="timestamp"))
    if args.min_size:
        parts.append(_.size >= parse_size(args.min_size))
    if args.tagged:
        # Map-key predicates don't push down; user_tags_count does.
        parts.append(_.user_tags_count > 0)
    if args.type:
        parts.append(_.element_type == args.type.upper())
    elif args.largest or args.min_size or args.ext:
        parts.append(_.element_type == "FILE")   # size questions are file questions
    if not parts:
        sys.exit("Pick at least one selector (--largest/--owner/--uid/--ext/"
                 "--changed-since/--min-size/--tagged); see --help.")
    pred = parts[0]
    for p in parts[1:]:
        pred = pred & p
    return pred
def main():
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[1],
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--under", default="/",
                    help="subtree to search (startswith on parent_path — see NOTES)")
    ap.add_argument("--largest", action="store_true",
                    help="sort by size; with no --min-size, implies --min-size 100M")
    ap.add_argument("--owner", help="owner_name exact match")
    ap.add_argument("--uid", type=int, help="numeric uid match")
    ap.add_argument("--ext", help="extension match (e.g. log, parquet)")
    ap.add_argument("--changed-since", help="mtime >= this UTC time ('YYYY-MM-DD [HH:MM]')")
    ap.add_argument("--min-size", help="size >= this (500M, 2G, ...)")
    ap.add_argument("--tagged", action="store_true",
                    help="only elements with S3 user tags (user_tags_count > 0)")
    ap.add_argument("--type", help="element_type (FILE, DIR, TABLE, ...); default FILE"
                                   "for size/extension questions")
    ap.add_argument("--limit", type=int, default=1000,
                    help="scan cap passed as limit_rows (default 1000)")
    ap.add_argument("--timeout", type=int, default=10, help="request timeout seconds")
    args = ap.parse_args()
    if args.largest and not args.min_size:
        args.min_size = "100M"
    columns = COLUMNS + (["user_tags"] if args.tagged else [])
    with connect(args.timeout).transaction() as tx:
        table = tx.catalog()
        df = table.select(columns=columns, predicate=build_predicate(args),
                          limit_rows=args.limit).read_all().to_pandas()
    if df.empty:
        print("No rows matched. Loosen a selector, or check --under against your "
              "tenant's namespace (see ../03-troubleshooting.md).")
        return
    df["path"] = df["parent_path"] + df["name"]
    df["size_h"] = df["size"].map(human)
    sort_key = "size" if (args.largest or args.min_size) else "mtime"
    show = ["path", "element_type", "size_h", "mtime", "owner_name"]
    if args.tagged:
        show.append("user_tags")
    out = df.sort_values(sort_key, ascending=False)[show]
    print(out.to_string(index=False))
    if len(df) == args.limit:
        print(f"\nNOTE: hit the --limit cap ({args.limit}): this is the sort of the first "
              f"{args.limit} matches the scan found, not a global top-N. Narrow the "
              "predicate (deeper --under, higher --min-size) or raise --limit.")
if __name__ == "__main__":
    main()
```

#### first_query.py

```python
#!/usr/bin/env python3
"""
first_query.py: first steps against the VAST Catalog with the vastdb SDK.
Opens the catalog by its built-in well-known name (tx.catalog()) and runs the
three things everyone wants first: list a directory, see the schema, and find
out which element types exist on this cluster. Verified on VAST 5.4.4.0
(vastdb 2.0.14).
SETUP
    pip install vastdb pyarrow pandas ibis-framework
    export VASTDB_ENDPOINT='http://<cluster-db-data-vip>'   # data VIP, not the VMS address
    export VASTDB_ACCESS_KEY='<access-key>'
    export VASTDB_SECRET_KEY='<secret-key>'
USAGE
    python first_query.py                      # list the namespace root
    python first_query.py --path /projects/    # list one directory (fast: equality)
    python first_query.py --types              # which element_type values exist here
    python first_query.py --schema             # print all catalog columns
NOTES
    Directory listing is an EQUALITY predicate on parent_path and needs the
    trailing slash ('/projects/', not '/projects') — without it you get zero
    rows. Expect seconds, not milliseconds: when a listing has fewer entries
    than --limit, the scan sweeps the whole catalog before returning (~15s on
    a 3B-row cluster). Always pass limit_rows and keep it close to what you
    need. See ../03-troubleshooting.md (Error 3).
"""
import argparse
import os
import sys
try:
    import vastdb
    from vastdb.config import BackoffConfig
    from ibis import _
except ImportError:
    sys.exit("Missing deps. Run: pip install vastdb pyarrow pandas ibis-framework")
LIST_COLUMNS = ["name", "element_type", "size", "used", "mtime", "owner_name", "uid"]
# Known element_type values across versions; --types probes which exist here.
ELEMENT_TYPE_CANDIDATES = ["FILE", "DIR", "SYMLINK", "TABLE", "SCHEMA", "BLOCK_DIR"]
def connect(timeout):
    endpoint = os.environ.get("VASTDB_ENDPOINT")
    access = os.environ.get("VASTDB_ACCESS_KEY")
    secret = os.environ.get("VASTDB_SECRET_KEY")
    if not (endpoint and access and secret):
        sys.exit("Set $VASTDB_ENDPOINT, $VASTDB_ACCESS_KEY, $VASTDB_SECRET_KEY.")
    # Fail fast on a wrong endpoint instead of retrying for a minute.
    return vastdb.connect(endpoint=endpoint, access=access, secret=secret,
                          timeout=timeout, backoff_config=BackoffConfig(max_tries=1))
def normalize_dir(path):
    if not path.endswith("/"):
        path += "/"
    return path
def main():
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[1],
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--path", default="/", help="directory to list (default: /)")
    ap.add_argument("--types", action="store_true",
                    help="probe which element_type values exist on this cluster")
    ap.add_argument("--schema", action="store_true", help="print the catalog columns")
    ap.add_argument("--limit", type=int, default=50, help="max rows to list (default 50)")
    ap.add_argument("--timeout", type=int, default=10, help="request timeout seconds")
    args = ap.parse_args()
    with connect(args.timeout).transaction() as tx:
        table = tx.catalog()   # the VAST Catalog, by its built-in well-known name
        if args.schema:
            for field in table.columns():
                print(f"{field.name}: {field.type}")
            return
        if args.types:
            # One cheap limit_rows=1 membership probe per candidate — never a full scan.
            present = [et for et in ELEMENT_TYPE_CANDIDATES
                       if table.select(columns=["name"],
                                       predicate=(_.element_type == et),
                                       limit_rows=1).read_all().num_rows]
            print("element_type values present on this cluster:")
            for et in present:
                print(f"  {et}")
            print("(candidate list is version-specific; yours may have more — "
                  "see ../04-schema-and-query-recipes.md)")
            return
        path = normalize_dir(args.path)
        df = table.select(columns=LIST_COLUMNS,
                          predicate=(_.parent_path == path),
                          limit_rows=args.limit).read_all().to_pandas()
        if df.empty:
            print(f"No entries under {path!r}. Check the trailing slash and that the "
                  "path is in your tenant's namespace; see ../03-troubleshooting.md.")
            return
        print(df.sort_values(["element_type", "name"]).to_string(index=False))
        if len(df) == args.limit:
            print(f"\n(showing the first {args.limit} entries; raise --limit for more)")
if __name__ == "__main__":
    main()
```

#### query_vms_catalog.py

```python
#!/usr/bin/env python3
"""
query_vms_catalog.py: query the VAST Catalog through the VMS REST API (vastpy).
Reads the catalog through the management plane, so it works with just VMS
credentials, as soon as the catalog is enabled. Verified on VAST 5.4.4.0.
(For the SDK path, see ../05-advanced-sdk-access.md.)
SETUP
    pip install vastpy
    export VMS_ADDRESS='var202.selab.vastdata.com'   # VMS/GUI host (bare), not the S3 endpoint
    export VMS_USER='admin'
    export VMS_PASSWORD='********'
    # or:  export VMS_TOKEN='<api-token>'   (VAST 5.3+)
USAGE
    python query_vms_catalog.py --path /projects/ --top 20
    python query_vms_catalog.py --path / --type TABLE --top 50
    python query_vms_catalog.py --path /home/ --name-prefix report --top 20
    python query_vms_catalog.py --path / --owner root --type DIR
NOTES
    The endpoint is bigcatalogconfig/query_data. It requires BOTH 'path' (the
    subtree to search, recursive) and 'filters' (at least one). Confirmed filter
    operators on 5.4.4: 'eq' and 'startswith'; 'contains' and range operators
    return 400 — check https://<vms>/docs/ for your version. element_type values
    are UPPERCASE (FILE, DIR, SYMLINK, TABLE, ...); list yours with
    `python first_query.py --types`.
"""
import argparse
import os
import sys
try:
    import urllib3
    urllib3.disable_warnings()
    from vastpy import VASTClient
except ImportError:
    sys.exit("Missing deps. Run: pip install vastpy")
DEFAULT_FIELDS = ["parent_path", "name", "element_type", "size", "used",
                  "mtime", "owner_name", "uid"]
def build_client():
    address = os.environ.get("VMS_ADDRESS")
    if not address:
        sys.exit("$VMS_ADDRESS is required (the VMS/GUI hostname, not the S3 data endpoint).")
    if address.startswith(("http://", "https://")):
        sys.exit("$VMS_ADDRESS must be a bare hostname; vastpy adds https:// itself.")
    kwargs = {"address": address}
    if os.environ.get("VMS_TOKEN"):
        kwargs["token"] = os.environ["VMS_TOKEN"]
    else:
        user, pwd = os.environ.get("VMS_USER"), os.environ.get("VMS_PASSWORD")
        if not (user and pwd):
            sys.exit("Set $VMS_USER and $VMS_PASSWORD (or $VMS_TOKEN).")
        kwargs["user"], kwargs["password"] = user, pwd
    if os.environ.get("VMS_TENANT_NAME"):
        kwargs["tenant_name"] = os.environ["VMS_TENANT_NAME"]
    return VASTClient(**kwargs)
def build_filters(args):
    filters = {}
    if args.type:
        filters["element_type"] = [{"eq": args.type.upper()}]
    if args.name:
        filters["name"] = [{"eq": args.name}]
    if args.name_prefix:
        filters["name"] = [{"startswith": args.name_prefix}]
    if args.owner:
        filters["owner_name"] = [{"eq": args.owner}]
    if not filters:
        # The endpoint requires filters, and there is no match-all operator
        # (an empty-prefix startswith is rejected with 400 too).
        sys.exit("Pass at least one of --type/--name/--name-prefix/--owner "
                 "(the endpoint requires a filter).")
    return filters
def normalize(resp):
    """query_data returns {'prop_list': [...], 'results': [[...], ...]}. Yield dict rows."""
    if not isinstance(resp, dict) or "results" not in resp:
        sys.exit(f"unexpected response: {resp!r}")
    cols = resp["prop_list"]
    for row in resp["results"]:
        yield dict(zip(cols, row))
def main():
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[1],
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--path", default="/",
                    help="subtree to search, recursive (default: / — the whole cluster;"
                         "this path reads through VMS and sees every tenant)")
    ap.add_argument("--type", help="element_type filter, e.g. FILE, DIR, TABLE, SYMLINK")
    ap.add_argument("--name", help="exact element name")
    ap.add_argument("--name-prefix", help="name starts with (operator: startswith)")
    ap.add_argument("--owner", help="owner_name exact match")
    ap.add_argument("--top", type=int, default=20, help="rows to fetch (default 20)")
    args = ap.parse_args()
    client = build_client()
    resp = client.bigcatalogconfig.query_data.post(
        path=args.path,
        limit=args.top,
        fields=DEFAULT_FIELDS,
        filters=build_filters(args),
    )
    rows = list(normalize(resp))
    if not rows:
        print("No rows matched. Widen --path or drop a filter; "
              "see ../03-troubleshooting.md.")
        return
    widths = {k: max(len(k), *(len(str(r.get(k, ""))) for r in rows)) for k in rows[0]}
    print("  ".join(k.ljust(widths[k]) for k in rows[0]))
    for r in rows:
        print("  ".join(str(r.get(k, "")).ljust(widths[k]) for k in rows[0]))
    count = resp.get("count")
    if count is not None:
        print(f"\n(count reported by VMS: {count} — may be capped on very large result sets)")
if __name__ == "__main__":
    main()
```