storage

storage

Storage backends for metrics data.

Classes:

Name	Description
JSONLinesStorage	Storage backend using JSON Lines format.

JSONLinesStorage

JSONLinesStorage(base_path: Path)

Storage backend using JSON Lines format.

Parameters:

Name	Type	Description	Default
base_path	Path	Base directory for storing metrics files	required

Methods:

Name	Description
get_record_by_id	Get a specific record by run ID.
list_files	List all metrics files.
read_records	Read metrics records from storage.
write_record	Write a metrics record to storage.

Source code in .venv/lib/python3.10/site-packages/panther/metrics/storage.py

def __init__(self, base_path: Path):
    """Initialize the storage backend.

    Args:
        base_path: Base directory for storing metrics files
    """
    self.base_path = Path(base_path)
    self.base_path.mkdir(parents=True, exist_ok=True)

get_record_by_id

get_record_by_id(run_id: str) -> dict[str, Any] | None

Get a specific record by run ID.

Parameters:

Name	Type	Description	Default
run_id	str	The run ID to search for	required

Returns:

Type	Description
dict[str, Any] | None	The matching record or None

Source code in .venv/lib/python3.10/site-packages/panther/metrics/storage.py

def get_record_by_id(self, run_id: str) -> dict[str, Any] | None:
    """Get a specific record by run ID.

    Args:
        run_id: The run ID to search for

    Returns:
        The matching record or None
    """
    # Search through all files
    files = sorted(self.base_path.glob("*.jsonl"), reverse=True)
    for filepath in files:
        records = self._read_file(filepath)
        for record in records:
            if record.get("run_id") == run_id:
                return record
    return None

list_files

list_files() -> list[Path]

List all metrics files.

Source code in .venv/lib/python3.10/site-packages/panther/metrics/storage.py

def list_files(self) -> list[Path]:
    """List all metrics files."""
    return sorted(self.base_path.glob("*.jsonl"), reverse=True)

read_records

read_records(
    date: str | None = None, limit: int | None = None
) -> list[dict[str, Any]]

Read metrics records from storage.

Parameters:

Name	Type	Description	Default
date	str | None	Optional date filter (YYYY-MM-DD format)	None
limit	int | None	Optional limit on number of records to return	None

Returns:

Type	Description
list[dict[str, Any]]	List of metrics records

Source code in .venv/lib/python3.10/site-packages/panther/metrics/storage.py

def read_records(
    self, date: str | None = None, limit: int | None = None
) -> list[dict[str, Any]]:
    """Read metrics records from storage.

    Args:
        date: Optional date filter (YYYY-MM-DD format)
        limit: Optional limit on number of records to return

    Returns:
        List of metrics records
    """
    records = []

    if date:
        # Read from specific date file
        filepath = self.base_path / f"{date}.jsonl"
        if filepath.exists():
            records.extend(self._read_file(filepath))
    else:
        # Read from all files, newest first
        files = sorted(self.base_path.glob("*.jsonl"), reverse=True)
        for filepath in files:
            records.extend(self._read_file(filepath))
            if limit and len(records) >= limit:
                break

    # Sort by timestamp, newest first
    records.sort(key=lambda x: x.get("timestamp", ""), reverse=True)

    if limit:
        records = records[:limit]

    return records

write_record

write_record(record: dict[str, Any]) -> None

Write a metrics record to storage.

Parameters:

Name	Type	Description	Default
record	dict[str, Any]	The metrics record to store	required

Source code in .venv/lib/python3.10/site-packages/panther/metrics/storage.py

def write_record(self, record: dict[str, Any]) -> None:
    """Write a metrics record to storage.

    Args:
        record: The metrics record to store
    """
    # Create filename based on current date
    now = datetime.now()
    filename = f"{now.strftime('%Y-%m-%d')}.jsonl"
    filepath = self.base_path / filename

    # Append the record as a JSON line
    with open(filepath, "a", encoding="utf-8") as f:
        json.dump(record, f, separators=(",", ":"))
        f.write("\n")