Skip to content

Public Data

Public Datasets

PublicData is a read-only list/get interface for the public S3 datasets that ship with Workbench.

The PublicData class lets you discover and pull down curated public datasets (e.g. computational chemistry benchmarks) without any AWS setup of your own.

Workbench REPL

Experiment with the PublicData() class in the Workbench REPL.

PublicData: Read-only access to public S3 data (comp_chem datasets)

PublicData

PublicData: Read-only list/get interface for public S3 datasets

Common Usage
public_data = PublicData()

# List available datasets
public_data.list()

# Get a specific dataset
df = public_data.get("comp_chem/aqsol/aqsol_public_data")
print(df)
Source code in src/workbench/api/public_data.py
class PublicData:
    """PublicData: Read-only list/get interface for public S3 datasets

    Common Usage:
        ```python
        public_data = PublicData()

        # List available datasets
        public_data.list()

        # Get a specific dataset
        df = public_data.get("comp_chem/aqsol/aqsol_public_data")
        print(df)
        ```
    """

    # Public bucket
    BUCKET = "workbench-public-data"

    def __init__(self):
        """PublicData Init Method"""
        self.log = logging.getLogger("workbench")

        # Anonymous boto3 session and config (no credentials needed for public data)
        self.boto3_session = boto3.Session(region_name="us-west-2")
        self.unsigned_config = Config(signature_version=UNSIGNED)
        self.s3_client = self.boto3_session.client("s3", config=self.unsigned_config)

    def list(self) -> list:
        """List all available datasets

        Returns:
            list: Dataset names (relative paths without extensions) available in the public store.
        """
        datasets = []
        paginator = self.s3_client.get_paginator("list_objects_v2")
        for page in paginator.paginate(Bucket=self.BUCKET):
            for obj in page.get("Contents", []):
                key = obj["Key"]
                if obj["Size"] == 0:
                    continue
                # Strip file extensions (.csv, .parquet, etc.)
                name = key
                for ext in (".parquet", ".csv", ".json"):
                    if name.endswith(ext):
                        name = name[: -len(ext)]
                        break
                datasets.append(name)

        return sorted(datasets)

    @not_found_returns_none
    def get(self, name: str) -> Union[pd.DataFrame, None]:
        """Retrieve a dataset by name

        Args:
            name (str): The dataset name (as returned by list()).

        Returns:
            pd.DataFrame: The retrieved DataFrame or None if not found.
        """
        readers = {".parquet": pd.read_parquet, ".csv": pd.read_csv}
        for ext, reader in readers.items():
            key = f"{name}{ext}"
            try:
                resp = self.s3_client.get_object(Bucket=self.BUCKET, Key=key)
                self.log.info(f"Reading s3://{self.BUCKET}/{key}...")
                return reader(BytesIO(resp["Body"].read()))
            except self.s3_client.exceptions.NoSuchKey:
                continue

        self.log.warning(f"Dataset '{name}' not found in public data store.")
        return None

    def details(self) -> pd.DataFrame:
        """Return detailed metadata for all datasets

        Returns:
            pd.DataFrame: DataFrame with name, size (MB), and modified date for each dataset.
        """
        rows = []
        paginator = self.s3_client.get_paginator("list_objects_v2")
        for page in paginator.paginate(Bucket=self.BUCKET):
            for obj in page.get("Contents", []):
                key = obj["Key"]
                if obj["Size"] == 0:
                    continue
                rows.append(
                    {
                        "name": key,
                        "size (MB)": round(obj["Size"] / (1024 * 1024), 2),
                        "modified": obj["LastModified"].strftime("%Y-%m-%d %H:%M:%S"),
                    }
                )

        return pd.DataFrame(rows) if rows else pd.DataFrame(columns=["name", "size (MB)", "modified"])

    def describe(self, name: str) -> Optional[dict]:
        """Return a description of a dataset including source references.

        Args:
            name: Dataset name (e.g. "comp_chem/logp/logp_all").

        Returns:
            dict with description, column info, references, etc., or None if not found.
        """
        # Load descriptions from S3 (cached after first call)
        if not hasattr(self, "_descriptions"):
            self._descriptions = self._load_descriptions()

        # Build candidate keys: exact, with extensions, and basename variants
        import posixpath

        basename = posixpath.basename(name)
        # Strip extension from basename if present
        stem = basename
        for ext in (".parquet", ".csv", ".json"):
            if stem.endswith(ext):
                stem = stem[: -len(ext)]
                break

        candidates = [name, basename, stem, f"{stem}.csv", f"{stem}.parquet", f"{basename}.csv", f"{basename}.parquet"]
        for key in candidates:
            if key in self._descriptions:
                return self._descriptions[key]

        self.log.info(f"No description found for '{name}'")
        return None

    def _load_descriptions(self) -> dict:
        """Load descriptions.json from S3."""
        s3_key = "descriptions.json"
        try:
            resp = self.s3_client.get_object(Bucket=self.BUCKET, Key=s3_key)
            return json.loads(resp["Body"].read().decode("utf-8"))
        except Exception as e:
            self.log.info(f"Could not load descriptions from s3://{self.BUCKET}/{s3_key}: {e}")
            return {}

    def __repr__(self):
        """Return a string representation of the PublicData object."""
        details_df = self.details()
        if details_df.empty:
            return "PublicData: No datasets found."

        max_name_len = details_df["name"].str.len().max() + 2
        details_df["name"] = details_df["name"].str.ljust(max_name_len)
        details_df["size (MB)"] = details_df["size (MB)"].apply(lambda x: f"{x:.2f} MB")
        details_df["modified"] = details_df["modified"].apply(lambda x: f" ({x})")
        return details_df.to_string(index=False, header=False)

__init__()

PublicData Init Method

Source code in src/workbench/api/public_data.py
def __init__(self):
    """PublicData Init Method"""
    self.log = logging.getLogger("workbench")

    # Anonymous boto3 session and config (no credentials needed for public data)
    self.boto3_session = boto3.Session(region_name="us-west-2")
    self.unsigned_config = Config(signature_version=UNSIGNED)
    self.s3_client = self.boto3_session.client("s3", config=self.unsigned_config)

__repr__()

Return a string representation of the PublicData object.

Source code in src/workbench/api/public_data.py
def __repr__(self):
    """Return a string representation of the PublicData object."""
    details_df = self.details()
    if details_df.empty:
        return "PublicData: No datasets found."

    max_name_len = details_df["name"].str.len().max() + 2
    details_df["name"] = details_df["name"].str.ljust(max_name_len)
    details_df["size (MB)"] = details_df["size (MB)"].apply(lambda x: f"{x:.2f} MB")
    details_df["modified"] = details_df["modified"].apply(lambda x: f" ({x})")
    return details_df.to_string(index=False, header=False)

describe(name)

Return a description of a dataset including source references.

Parameters:

Name Type Description Default
name str

Dataset name (e.g. "comp_chem/logp/logp_all").

required

Returns:

Type Description
Optional[dict]

dict with description, column info, references, etc., or None if not found.

Source code in src/workbench/api/public_data.py
def describe(self, name: str) -> Optional[dict]:
    """Return a description of a dataset including source references.

    Args:
        name: Dataset name (e.g. "comp_chem/logp/logp_all").

    Returns:
        dict with description, column info, references, etc., or None if not found.
    """
    # Load descriptions from S3 (cached after first call)
    if not hasattr(self, "_descriptions"):
        self._descriptions = self._load_descriptions()

    # Build candidate keys: exact, with extensions, and basename variants
    import posixpath

    basename = posixpath.basename(name)
    # Strip extension from basename if present
    stem = basename
    for ext in (".parquet", ".csv", ".json"):
        if stem.endswith(ext):
            stem = stem[: -len(ext)]
            break

    candidates = [name, basename, stem, f"{stem}.csv", f"{stem}.parquet", f"{basename}.csv", f"{basename}.parquet"]
    for key in candidates:
        if key in self._descriptions:
            return self._descriptions[key]

    self.log.info(f"No description found for '{name}'")
    return None

details()

Return detailed metadata for all datasets

Returns:

Type Description
DataFrame

pd.DataFrame: DataFrame with name, size (MB), and modified date for each dataset.

Source code in src/workbench/api/public_data.py
def details(self) -> pd.DataFrame:
    """Return detailed metadata for all datasets

    Returns:
        pd.DataFrame: DataFrame with name, size (MB), and modified date for each dataset.
    """
    rows = []
    paginator = self.s3_client.get_paginator("list_objects_v2")
    for page in paginator.paginate(Bucket=self.BUCKET):
        for obj in page.get("Contents", []):
            key = obj["Key"]
            if obj["Size"] == 0:
                continue
            rows.append(
                {
                    "name": key,
                    "size (MB)": round(obj["Size"] / (1024 * 1024), 2),
                    "modified": obj["LastModified"].strftime("%Y-%m-%d %H:%M:%S"),
                }
            )

    return pd.DataFrame(rows) if rows else pd.DataFrame(columns=["name", "size (MB)", "modified"])

get(name)

Retrieve a dataset by name

Parameters:

Name Type Description Default
name str

The dataset name (as returned by list()).

required

Returns:

Type Description
Union[DataFrame, None]

pd.DataFrame: The retrieved DataFrame or None if not found.

Source code in src/workbench/api/public_data.py
@not_found_returns_none
def get(self, name: str) -> Union[pd.DataFrame, None]:
    """Retrieve a dataset by name

    Args:
        name (str): The dataset name (as returned by list()).

    Returns:
        pd.DataFrame: The retrieved DataFrame or None if not found.
    """
    readers = {".parquet": pd.read_parquet, ".csv": pd.read_csv}
    for ext, reader in readers.items():
        key = f"{name}{ext}"
        try:
            resp = self.s3_client.get_object(Bucket=self.BUCKET, Key=key)
            self.log.info(f"Reading s3://{self.BUCKET}/{key}...")
            return reader(BytesIO(resp["Body"].read()))
        except self.s3_client.exceptions.NoSuchKey:
            continue

    self.log.warning(f"Dataset '{name}' not found in public data store.")
    return None

list()

List all available datasets

Returns:

Name Type Description
list list

Dataset names (relative paths without extensions) available in the public store.

Source code in src/workbench/api/public_data.py
def list(self) -> list:
    """List all available datasets

    Returns:
        list: Dataset names (relative paths without extensions) available in the public store.
    """
    datasets = []
    paginator = self.s3_client.get_paginator("list_objects_v2")
    for page in paginator.paginate(Bucket=self.BUCKET):
        for obj in page.get("Contents", []):
            key = obj["Key"]
            if obj["Size"] == 0:
                continue
            # Strip file extensions (.csv, .parquet, etc.)
            name = key
            for ext in (".parquet", ".csv", ".json"):
                if name.endswith(ext):
                    name = name[: -len(ext)]
                    break
            datasets.append(name)

    return sorted(datasets)