diff --git a/studio/studio/helpers.py b/studio/studio/helpers/__init__.py similarity index 73% rename from studio/studio/helpers.py rename to studio/studio/helpers/__init__.py index 9d0eef89d..74edd8223 100644 --- a/studio/studio/helpers.py +++ b/studio/studio/helpers/__init__.py @@ -1,12 +1,27 @@ +""" +Shared UI helper functions for the Studio package. +Provides asset loading, rendering, validation, and session-state management. +""" + import os +import json +from typing import Dict, Any, Type, List, Callable, Union import streamlit as st -from typing import Dict, Any, Type, List, Callable +import pandas as pd from pydantic import BaseModel, ValidationError -import json -from promptops.utils import ROOT, iter_prompt_files, iter_workflow_files, load_yaml, save_yaml +from promptops.utils import ( + ROOT, + iter_prompt_files, + iter_workflow_files, + load_yaml, + save_yaml, +) + -def get_relative_asset_paths(asset_type: str, extensions: List[str] = None) -> list[str]: +def get_relative_asset_paths( + asset_type: str, extensions: List[str] = None +) -> list[str]: """ Scans the workspace for prompt or workflow files, returning a sorted list of relative paths. Conforms to the standard workspace traverser. @@ -18,13 +33,16 @@ def get_relative_asset_paths(asset_type: str, extensions: List[str] = None) -> l files = list(iter_workflow_files()) else: raise ValueError(f"Unknown asset type: {asset_type}") - + if extensions: files = [f for f in files if any(str(f).endswith(ext) for ext in extensions)] - + return sorted([os.path.relpath(str(f), base_dir) for f in files]) -def render_file_selector(asset_type: str, key: str, label: str = None, extensions: List[str] = None) -> str: + +def render_file_selector( + asset_type: str, key: str, label: str = None, extensions: List[str] = None +) -> str: """ Renders a unified selectbox for prompt or workflow files using relative paths. """ @@ -33,13 +51,14 @@ def render_file_selector(asset_type: str, key: str, label: str = None, extension label = f"Select a {asset_type.lower()} file" return st.selectbox(label, options=paths, key=key) + def render_schema_form( schema_class: Type[BaseModel], data: Dict[str, Any], skip_fields: List[str] = None, key_prefix: str = "", layout_config: Dict[str, List[str]] = None, - layout_type: str = "tabs" + layout_type: str = "tabs", ) -> Dict[str, Any]: """ Dynamically generates Streamlit inputs from a Pydantic model's JSON schema. @@ -47,13 +66,18 @@ def render_schema_form( """ if skip_fields is None: skip_fields = [] - + schema = schema_class.model_json_schema() properties = schema.get("properties", {}) - - def render_field(field_name: str, field_info: Dict[str, Any], data_dict: Dict[str, Any]) -> Any: + + def render_field( + field_name: str, field_info: Dict[str, Any], data_dict: Dict[str, Any] + ) -> Any: + """ + Renders a single schema field using appropriate Streamlit input elements. + """ val = data_dict.get(field_name, "") - + # Determine the label to use if "title" in field_info: label = field_info["title"] @@ -61,38 +85,52 @@ def render_field(field_name: str, field_info: Dict[str, Any], data_dict: Dict[st label = field_info["description"] else: label = field_name - + if field_name == "metadata": st.subheader("Metadata") if "metadata" not in data_dict or not data_dict["metadata"]: data_dict["metadata"] = {} - + meta_schema = {} if "$defs" in schema: for def_name, def_schema in schema["$defs"].items(): if "Metadata" in def_name: meta_schema = def_schema.get("properties", {}) break - + for m_key, m_info in meta_schema.items(): m_val = data_dict["metadata"].get(m_key, m_info.get("default", "")) m_label = m_info.get("title", m_key) if m_info.get("type") == "boolean": - data_dict["metadata"][m_key] = st.checkbox(m_label, value=bool(m_val), key=f"{key_prefix}meta_{m_key}") + data_dict["metadata"][m_key] = st.checkbox( + m_label, value=bool(m_val), key=f"{key_prefix}meta_{m_key}" + ) elif m_info.get("type") == "array": m_val_str = ", ".join(m_val) if isinstance(m_val, list) else "" - res = st.text_input(m_label, value=m_val_str, key=f"{key_prefix}meta_{m_key}") - data_dict["metadata"][m_key] = [x.strip() for x in res.split(",") if x.strip()] + res = st.text_input( + m_label, value=m_val_str, key=f"{key_prefix}meta_{m_key}" + ) + data_dict["metadata"][m_key] = [ + x.strip() for x in res.split(",") if x.strip() + ] else: - data_dict["metadata"][m_key] = st.text_input(m_label, value=m_val, key=f"{key_prefix}meta_{m_key}") + data_dict["metadata"][m_key] = st.text_input( + m_label, value=m_val, key=f"{key_prefix}meta_{m_key}" + ) elif field_info.get("type") == "string": if field_name == "description" or "description" in field_name.lower(): - data_dict[field_name] = st.text_area(label, value=val, key=f"{key_prefix}{field_name}") + data_dict[field_name] = st.text_area( + label, value=val, key=f"{key_prefix}{field_name}" + ) else: - data_dict[field_name] = st.text_input(label, value=val, key=f"{key_prefix}{field_name}") + data_dict[field_name] = st.text_input( + label, value=val, key=f"{key_prefix}{field_name}" + ) elif field_info.get("type") == "boolean": - data_dict[field_name] = st.checkbox(label, value=bool(val), key=f"{key_prefix}{field_name}") - + data_dict[field_name] = st.checkbox( + label, value=bool(val), key=f"{key_prefix}{field_name}" + ) + return data_dict if layout_config: @@ -104,7 +142,7 @@ def render_field(field_name: str, field_info: Dict[str, Any], data_dict: Dict[st if field_name in skip_fields or field_name not in properties: continue data = render_field(field_name, properties[field_name], data) - else: # collapsible / expanders + else: # collapsible / expanders for section_title, fields in layout_config.items(): with st.expander(section_title, expanded=True): for field_name in fields: @@ -116,11 +154,14 @@ def render_field(field_name: str, field_info: Dict[str, Any], data_dict: Dict[st if field_name in skip_fields: continue data = render_field(field_name, field_info, data) - + return data + def load_asset_data(file_path: str, raw: bool = True) -> Dict[str, Any]: - """Loads and returns YAML data. Standardizes error displays.""" + """ + Loads and returns YAML data. Standardizes error displays. + """ try: if os.path.exists(file_path): return load_yaml(file_path, raw=raw) or {} @@ -128,12 +169,13 @@ def load_asset_data(file_path: str, raw: bool = True) -> Dict[str, Any]: st.error(f"Failed to load file: {e}") return {} + def validate_and_save_asset( file_path: str, data: Dict[str, Any], schema_class: Type[BaseModel], save_callback: Callable = None, - success_message: str = None + success_message: str = None, ) -> bool: """ Validates data against the Pydantic schema_class. @@ -143,11 +185,15 @@ def validate_and_save_asset( # 1. Path safety and traversal checks before any disk/folder operations base_dir = str(ROOT) if has_path_traversal(file_path): - st.error("Path validation failed: directory traversal segments ('..') are not allowed.") + st.error( + "Path validation failed: directory traversal segments ('..') are not allowed." + ) return False if not is_safe_path(base_dir, file_path): - st.error("Path validation failed: target path is outside the allowed workspace boundaries.") + st.error( + "Path validation failed: target path is outside the allowed workspace boundaries." + ) return False # 2. Safe directory creation AFTER successful path validation @@ -162,14 +208,17 @@ def validate_and_save_asset( try: # Validate using Pydantic model instantiation schema_class(**data) - + # Save operation if save_callback: save_callback(file_path, data) else: save_yaml(file_path, data) - - msg = success_message or f"Successfully saved and validated {os.path.basename(file_path)}!" + + msg = ( + success_message + or f"Successfully saved and validated {os.path.basename(file_path)}!" + ) st.success(msg) return True except json.JSONDecodeError as e: @@ -183,9 +232,12 @@ def validate_and_save_asset( return False -def _get_all_matching_templates(indices: List[int], key_patterns: List[str]) -> List[str]: - """Finds all key templates currently present in session state for any of the given indices.""" - import streamlit as st +def _get_all_matching_templates( + indices: List[int], key_patterns: List[str] +) -> List[str]: + """ + Finds all key templates currently present in session state for any of the given indices. + """ templates = set() for idx in indices: for key in list(st.session_state.keys()): @@ -195,33 +247,36 @@ def _get_all_matching_templates(indices: List[int], key_patterns: List[str]) -> if key.startswith(prefix_part): expected_prefix = prefix_part + str(idx) if key.startswith(expected_prefix): - dynamic_suffix = key[len(expected_prefix):] + dynamic_suffix = key[len(expected_prefix) :] # Make sure the dynamic suffix matches suffix_part if there is one - if suffix_part == "" or dynamic_suffix.startswith(suffix_part): + if suffix_part == "" or dynamic_suffix.startswith( + suffix_part + ): templates.add(prefix_part + "{}" + dynamic_suffix) else: # Pattern is a simple prefix if key.startswith(f"{pattern}{idx}_") or key == f"{pattern}{idx}": - suffix = key[len(f"{pattern}{idx}"):] + suffix = key[len(f"{pattern}{idx}") :] templates.add(f"{pattern}{{}}{suffix}") return list(templates) def _swap_session_keys(i: int, j: int, key_patterns: List[str]): - """Swaps keys in st.session_state matching format patterns for indices i and j.""" - import streamlit as st + """ + Swaps keys in st.session_state matching format patterns for indices i and j. + """ templates = _get_all_matching_templates([i, j], key_patterns) for template in templates: key_i = template.format(i) key_j = template.format(j) val_i = st.session_state.get(key_i, None) val_j = st.session_state.get(key_j, None) - + if val_j is not None: st.session_state[key_i] = val_j elif key_i in st.session_state: del st.session_state[key_i] - + if val_i is not None: st.session_state[key_j] = val_i elif key_j in st.session_state: @@ -229,11 +284,12 @@ def _swap_session_keys(i: int, j: int, key_patterns: List[str]): def _delete_session_keys(i: int, length: int, key_patterns: List[str]): - """Shifts keys matching format patterns down when item at index i is deleted.""" - import streamlit as st + """ + Shifts keys matching format patterns down when item at index i is deleted. + """ # Find all templates for all possible indices in the list templates = _get_all_matching_templates(list(range(length)), key_patterns) - + for idx in range(i, length - 1): for template in templates: key_curr = template.format(idx) @@ -242,7 +298,7 @@ def _delete_session_keys(i: int, length: int, key_patterns: List[str]): st.session_state[key_curr] = st.session_state[key_next] elif key_curr in st.session_state: del st.session_state[key_curr] - + last_idx = length - 1 for template in templates: key_last = template.format(last_idx) @@ -255,19 +311,18 @@ def render_shared_list( item_renderer: Callable[[int, Any], None], key_patterns: List[str] = None, col_widths: List[float] = [8.5, 1.5], - show_reorder: bool = True + show_reorder: bool = True, ) -> None: """ Renders a unified sequence of list items with standard controls: - Move Up (disabled for the first item) - Move Down (disabled for the last item) - Delete (standard delete button across editors) - + Ensures identical layout ratio and standard Streamlit styling. Strictly manages order changes and deletion, leaving content editing/rendering to the item_renderer. Automatically handles session state synchronization and triggers rerun. """ - import streamlit as st items = st.session_state.get(session_state_key, []) if not items: return @@ -278,46 +333,59 @@ def render_shared_list( for i in range(len(items)): # Render standard columns layout for identical layout ratios across both editors col_content, col_actions = st.columns(col_widths) - + with col_content: item_renderer(i, items[i]) - + with col_actions: # Standard vertical spacing - st.write("") # creates slight top alignment spacing - + st.write("") # creates slight top alignment spacing + if show_reorder: - up_disabled = (i == 0) - down_disabled = (i == len(items) - 1) - + up_disabled = i == 0 + down_disabled = i == len(items) - 1 + # Action buttons use unified, consistent plain-text labels and secondary button styles - if st.button("⬆️ Move Up", key=f"move_up_{session_state_key}_{i}", disabled=up_disabled, use_container_width=True): - items[i], items[i-1] = items[i-1], items[i] + if st.button( + "⬆️ Move Up", + key=f"move_up_{session_state_key}_{i}", + disabled=up_disabled, + use_container_width=True, + ): + items[i], items[i - 1] = items[i - 1], items[i] _swap_session_keys(i, i - 1, key_patterns) st.rerun() - - if st.button("⬇️ Move Down", key=f"move_down_{session_state_key}_{i}", disabled=down_disabled, use_container_width=True): - items[i], items[i+1] = items[i+1], items[i] + + if st.button( + "⬇️ Move Down", + key=f"move_down_{session_state_key}_{i}", + disabled=down_disabled, + use_container_width=True, + ): + items[i], items[i + 1] = items[i + 1], items[i] _swap_session_keys(i, i + 1, key_patterns) st.rerun() - - if st.button("🗑️ Delete", key=f"delete_{session_state_key}_{i}", use_container_width=True): + + if st.button( + "🗑️ Delete", + key=f"delete_{session_state_key}_{i}", + use_container_width=True, + ): items.pop(i) _delete_session_keys(i, len(items) + 1, key_patterns) st.rerun() -import pandas as pd -from typing import List, Dict, Any, Union - -def sanitize_dataframe_records(data: Union[pd.DataFrame, List[Dict[str, Any]]], boolean_cols: List[str] = None) -> List[Dict[str, Any]]: +def sanitize_dataframe_records( + data: Union[pd.DataFrame, List[Dict[str, Any]]], boolean_cols: List[str] = None +) -> List[Dict[str, Any]]: """ Sanitizes UI dataframes or list of dicts before validation and saving. - Converts NaN/pd.NA and empty or whitespace-only strings to None. - Converts empty or unset checkbox fields (in boolean_cols) to False instead of NaN. """ if isinstance(data, pd.DataFrame): - records = data.to_dict('records') + records = data.to_dict("records") elif isinstance(data, list): # Shallow copy to avoid mutation records = [dict(r) for r in data] @@ -325,16 +393,21 @@ def sanitize_dataframe_records(data: Union[pd.DataFrame, List[Dict[str, Any]]], return data if boolean_cols is None: - boolean_cols = ['required'] + boolean_cols = ["required"] sanitized = [] for row in records: clean_row = {} for k, v in row.items(): is_null = pd.isna(v) if not isinstance(v, (list, dict, str)) else False - + if k in boolean_cols: - if is_null or v is None or v == "" or (isinstance(v, str) and v.strip() == ""): + if ( + is_null + or v is None + or v == "" + or (isinstance(v, str) and v.strip() == "") + ): clean_row[k] = False else: clean_row[k] = bool(v) @@ -350,11 +423,11 @@ def sanitize_dataframe_records(data: Union[pd.DataFrame, List[Dict[str, Any]]], def has_path_traversal(path_str: str) -> bool: - """ + r""" Checks if a path contains directory traversal sequences like '..', '../', '..\\'. """ - normalized = path_str.replace('\\', '/') - segments = normalized.split('/') + normalized = path_str.replace("\\", "/") + segments = normalized.split("/") if ".." in segments: return True if ".." in path_str: @@ -382,13 +455,10 @@ def get_existing_subfolders(asset_root_dir: str) -> List[str]: subfolders = [] if os.path.exists(asset_root_dir): for root, dirs, _ in os.walk(asset_root_dir): - dirs[:] = [d for d in dirs if not d.startswith('.') and d != "__pycache__"] + dirs[:] = [d for d in dirs if not d.startswith(".") and d != "__pycache__"] for d in dirs: full_path = os.path.join(root, d) rel_path = os.path.relpath(full_path, asset_root_dir) if rel_path and rel_path != ".": subfolders.append(rel_path) return ["."] + sorted(list(set(subfolders))) - - - diff --git a/studio/studio/helpers/ruff.toml b/studio/studio/helpers/ruff.toml new file mode 100644 index 000000000..c47a0332e --- /dev/null +++ b/studio/studio/helpers/ruff.toml @@ -0,0 +1,19 @@ +extend = "../../ruff.toml" + +[lint] +select = [ + "E", # pycodestyle errors + "W", # pycodestyle warnings + "F", # pyflakes + "D", # pydocstyle +] +ignore = [ + "D200", # One-line docstring should fit on one line + "D205", # 1 blank line required between summary line and description + "D212", # Multi-line docstring summary should start at the first line + "D415", # First line should end with a period, question mark, or exclamation point + "E501", # Line too long (handled by formatting) +] + +[lint.pydocstyle] +convention = "google" diff --git a/tools/tools/scripts/check_docstrings.py b/tools/tools/scripts/check_docstrings.py index 4dda7c61b..d1f3f3e65 100644 --- a/tools/tools/scripts/check_docstrings.py +++ b/tools/tools/scripts/check_docstrings.py @@ -45,7 +45,8 @@ def main(): """Missing docstring.""" directories = [ "promptops/promptops", - "tools/tools/scripts" + "tools/tools/scripts", + "studio/studio/helpers" ] passed = True