Streamlit Handler Class
The class is to help with creating the Streamlit widgets and communication between the widgets and the databasee.
Three classes inherit from the StreamlitHandler class
- StreamlitAddData
- StreamlitEditData
- StreamlitDeleteData
The three classes as you can guess relates to the app’s corresponding page.
StreamlitHandler – Base
I won’t go through all of the attributes, most are self explaintory. However, the one thing to mention is that the class uses two other classes, the PandasSQLHandler and the DataHandler class. We will discuss those classes further on in the tutorial.
Methods
_load_query
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
def _load_query(self, query_file_name:str = None): """ Read the query file stored in /query directory Name of file needs to match the table name which is displayed in the drop down box Parameters ---------- query_file_name : str, optional name of query file to be loaded in the class if the query name is not specified, it is assumed to use a file name of schema.table_name.sql which is assigned in the source_table_name property. """ # if user supplied query_file_name else default val from class query_file_name = query_file_name if query_file_name is not None else self._query_file_name with open(f"query/{query_file_name}", 'r') as file: self.query = file.read() |
Read the query file stored in /query directory. Name of file needs to match the table name which is displayed in the drop down box. The names of the tables to be displayed are configured in the config / sql_column_config.yaml file.
__set_primary_key_column_name
|
1 2 3 4 5 6 |
def __set_primary_key_column_name(self) -> None: """ sets the primary key column name defined in the config/sql_column_config.yaml file """ self.primary_key_column_name = sql_column_config.table[self._base_table_name].PRIMARY_KEY |
sets the primary key column name defined in the config/sql_column_config.yaml file.
__set_column_order
|
1 2 3 4 5 6 7 8 9 10 11 |
def __set_column_order(self) -> None: """ sets the column order for the data_editor widgets, while excluding the primary key column. values are defined in the config/sql_column_config.yaml file """ table_columns = list(self.table_data.columns) if self.primary_key_column_name in table_columns: table_columns.remove(self.primary_key_column_name) self._column_order = tuple(table_columns) |
sets the column order for the data_editor widgets, while excluding the primary key column. The values and order are defined in the config/sql_column_config.yaml file. The data_editor widgets will display the columns in the same order as defined in the configuration file.
fetch_row_count
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
def fetch_row_count(self) -> int: """ Fetch number of rows in the table, used in pagination Returns ------- int """ # load the query string self._load_query(sql_column_config.info.count.QUERY) # fetch the row_count data data = self.db_handler.read_table(query=self.query, params=(self.source_table_name,) ) return data.loc[0, 'row_count'] |
Fetch number of rows in the table, used in pagination.
Will run a quick query, The query to use is defined in the config/sql_column_config.yaml file. In our tutorial we are connecting to Microsoft SQL database. Instead of doing a row_count query which mayy be expensive, will read from the sys.dm_db_partition_stats table which doesn’t require elevated privileges. If the user has access to read / write to tables they should be able to read from sys.dm_db_partition_stats.
fetch_table_data
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
def fetch_table_data(self,row_params:tuple=None) -> None: """Loads the data from source table Returns ------- None """ self.__set_primary_key_column_name() self._load_query() self.table_data = self.db_handler.read_table(self.query, row_params) # set the column order, hiding the primary key column self.__set_column_order() |
Method will query the database, using the PandasSQLHandler class. Storing the results to be sent to the data_editor widget.
jump_current_page
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
def jump_current_page(self)-> None: """ When user increments, decrements, enters in page number move to that page """ page_input = st.session_state['current_page_input'] if page_input >= 1 and page_input <= self._total_pages: self.current_page = page_input self.fetch_table_data(row_params=((self.current_page-1)*self.rows_per_page, self.rows_per_page, )) |
Will query the source table, inclusive of prior database commits and based on user selected number of rows to display and total rows in the table will jump to the specified page. The requested page is read from Streamlist session_state dictionary. Conditional logic is in place to prevent from requesting a page that doesn’t exist.
move_current_page
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
def move_current_page(self, action:str)-> None: """ When user increments, decrements, enters in page number move to that page """ match action: case 'decrease': if self.current_page > 1: self.current_page -= 1 case 'increase': if self.current_page < self._total_pages: self.current_page += 1 case _: self.current_page = 1 self.fetch_table_data(row_params=((self.current_page-1)*self.rows_per_page, self.rows_per_page, )) |
Will query the source table, inclusive of prior database commits and based on user selected number of rows to display and total rows in the table will move back or forward one page. Using call backs from the nav control buttons, determines the user requested action. Conditional logic is in place to prevent from requesting a page that doesn’t exist.
input_page_select
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
def input_page_select(self): """ Creates the streamlit input UI element, limiting to number of pages Returns ------- streamlit.selectbox """ # make sure we don't exceed max number of pages if self.current_page >= self.total_pages: self.current_page = int(self.total_pages) return(st.number_input( "Go to page", min_value=1, max_value=self.total_pages, value=self.current_page, step=1, key="current_page_input", on_change=self.jump_current_page, )) |
Returns a Streamlit number_input widget. The widget allows user to input which page number they want to jump too. The widget is used in the page navigation controls show in the sidebar.
page_nav_controls
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 |
def page_nav_controls(self): """ Creates the streamlit input UI elments for page navigation Returns ------- streamlit.selectbox """ with st.container(key='page-controller', border=True): col1, col2, col3, col4, col5 = st.columns([1, 2, 1, 1, 4], vertical_alignment="bottom" ) with col1: page_nav_label = st.markdown('<p style="font-size:1.3rem;">Page</p>', unsafe_allow_html=True) with col2: page_nav_page_info = st.markdown(f'<p style="font-size:1.3rem;">{self.current_page} of {self.total_pages}</p>', unsafe_allow_html=True) with col3: page_nav_prior_page = st.button(label="<<", type="secondary", disabled=(self.current_page <= 1), on_click=self.move_current_page, args=('decrease',) ) with col4: page_nav_next_page = st.button(label=">>", type="secondary", disabled=(self.current_page >= self.total_pages), on_click=self.move_current_page, args=('increase',) ) with col5: # intentionally left the column empty page_nav_empty = st.empty() # just a spacer column # return the page_nav controls return (page_nav_label, page_nav_page_info, page_nav_prior_page, page_nav_next_page, page_nav_empty ) |
The method will return five streamlit widgets that help the user navigate between pages in the data_editor widget. I used st.markdown widget to pass the label and in the current page information. The other benefit is that you can pass css style properties. You do have to use the unsafe_allow_html property to True, for the Streamlit to apply the css style properties.
The buttons will navigate to prior page or next page, and use a call back to move_current_page method.
selectbox_rows_per_page
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
def selectbox_rows_per_page(self): """ Creates the streamlit drop down UI element, based on the table names stored in the /config/sql_column_config.yaml file Returns ------- streamlit.selectbox """ return (st.selectbox(label="Rows per Page:", options=streamlit_config.selectbox.rows_per_page.OPTIONS, index=0, on_change=self.update_rows_per_page, key="rows_per_page", )) |
Returns a select box that allows the user to determine how many rows to show in the data_editor widgets. The values are defined in the config / streamlit_config.yaml file.
selectbox_table
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 |
def selectbox_table(self): """ Creates the streamlit drop down UI element, based on the table names stored in the /config/sql_column_config.yaml file Returns ------- streamlit.selectbox """ table_options = [f"{sql_column_config.table[tbl].SCHEMA}.{sql_column_config.table[tbl].NAME}" for tbl in sql_column_config.table] with st.container(): col_left, col_mid, col_right = st.columns([4,4,4]) with col_left: select_box_table = st.selectbox(label="Select a table to edit:", options=table_options, index=None, placeholder='Select a Table', ) with col_mid: # intentionally left the column empty st.empty() with col_right: # intentionally left the column empty st.empty() return (select_box_table) |
Returns a streamlit select_box widget, listing the schema + table name to the user. The schema and table names are listed in the config / sql_column_config.yaml file. The select_box widget is wrapped in a container allowing for formatting the width on the page. Default behavior is for the select_box widget to use the full width of the page.
query_success
|
1 2 3 4 5 6 7 8 |
def query_success(self): """ Returns ------- boolean indicates if the last query run was completed successfully """ return(self._query_success) |
Returns a Boolean, indicates if the last query run was completed successfully.
update_rows_per_page
|
1 2 3 4 5 |
def update_rows_per_page(self) -> None: """ Update number of rows per page to return """ self.rows_per_page = st.session_state['rows_per_page'] |
Reads from the streamlit sesssion_state dictionary, number of rows per page to return.
StreamlitAddData
The class inherits from the StreamlitHandler class with additional methods for adding rows to the table.
Methods
fetch_empty_table
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 |
def fetch_empty_table(self, table_name:str = None): """ Create an empty data_editor widget with appropriate columns for user to input data to be inserted into the source table Parameters ---------- table_name : str Name of source table Returns ------- streamlit data_editor """ query_params = (self._source_schema, self._base_table_name,) column_info = self.data_handler.get_table_columns( query=self.query, filter_columns=sql_column_config.table[self._base_table_name].COLUMNS, query_params=query_params ) data = pd.DataFrame({column_name: pd.Series(dtype=column_type) for column_name, column_type in column_info.items()}) return(st.data_editor(data=data, num_rows="dynamic", use_container_width=True )) |
Create the streamlit data_editor widget. However we need to know which columns and their corresponding datatypes to add to the data_editor widget. In order to do so wee will query the meta data of the table using the get_table_columns method from the data_handler class. The method will also align the data types from Microsoft SQL to pandas dataframe datatypes. Additionally we will only show to the user, the columns specified in the config file, config / sql_column_config.yaml
add_source_data
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
def add_source_data(self, new_data: pd.DataFrame): """ Add new rows to the source Parameters ---------- table_name : pandas DataFrame DataFrame will hold data to be inserted in to the source table """ for row_ctr in range(0, new_data.shape[0]): self.data_handler.add_row(self.source_table_name, new_data.loc[[row_ctr]] ) self._query_success = True |
Now, we want to add the user input as a row to the table. The data will be passed from the data_editor widget to thhe database using the data_handler add_row method. We loop through the data_editor rows because the user may try adding multiple rows at the same time. If all of the rows are added successfully we set the _query_success flag to True. The flag is used by the query_success method.
StreamlitEditData
The class inherits from the StreamlitHandler class with additional methods for editing rows in the table.
Methods
data_editor_formatted
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 |
def data_editor_formatted(self): """ Will return a formatted data_editor widget Returns ------- streamlit data_editor widget """ # offset = (current_page - 1) * rows per page # fetch next = rows per page self.fetch_table_data(row_params=((self.current_page-1)*self.rows_per_page, self.rows_per_page, )) return(st.data_editor(data=self.table_data, key="data_editor_update", use_container_width=True, column_order=self._column_order, )) |
Using the fetch_table_data method will fetch the data from the source table. Will pass the offset and number of rows to return to the query to the row_params parameter. The values have to be passed as a tuple.
Return the streamlit data_editor widget to the user.
update_source_data
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 |
def update_source_data(self, edited_data: pd.DataFrame)-> None: """ Detect changes by comparing the original and edited DataFrames Parameters ---------- edited_data : pandas DataFrame Will extract out the rows and columns that need to be updated """ if not edited_data.equals(self.table_data): # get what has changed changes = edited_data.compare(self.table_data) for index, row in changes.iterrows(): # Update only the changed columns for each modified row row_id = self.table_data.loc[index, self.primary_key_column_name] changed_columns = row.index.get_level_values(0).unique() # Extract only the changed values to update in the database update_data = edited_data.loc[index, changed_columns].to_dict() self.data_handler.update_row(table=self._remote_table_name, update_data=update_data, primary_key_column=self.primary_key_column_name, row_id=row_id) self._query_success = True |
First we do a comparison to the source data and the data_editor widget data. Both sets are data are stored in a Pandas data frame allowing us to quickly check if they are equal.
If they are not equal, we grab the changes by using the data frame .compare method.
Next we loop through the rows and columns to determine what has changed. We only want to update the column data that has changed. We store the changes in the update_data in a dictionary variable.
Lastly we call the data_handler update_row method to update the source table. The method allows for sending multiple row updates. However each row that has changed will create an UPDATE sql statement to be executed.
Lastly if all the updates have been process the we set the _query_success flag to True. The flag is used by the query_success method.
StreamlitDeleteData
The class inherits from the StreamlitHandler class with additional methods for editing rows in the table.
Although in the tutorial I am showing how to delete the rows, I would suggest only deleting table rows when you are keeping table history. If don’t have temporal (system-versioned) tables I would suggest the table structure uses a delete column to mark the row as deleted. And you an update query to change the column value.
Methods
data_editor_formatted
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 |
""" Will return a formatted data_editor widget Returns ------- streamlit data_editor widget """ # get the source table data self.fetch_table_data(row_params=((self.current_page-1)*self.rows_per_page, self.rows_per_page, )) # add column to data set self.table_data["Select"] = False # set the column order self._column_order = ('Select',) + self._column_order return(st.data_editor(data=self.table_data, key="data_editor_delete", use_container_width=True, column_order=self._column_order, )) |
Using the fetch_table_data method will fetch the data from the source table. Will pass the offset and number of rows to return to the query to the row_params parameter. The values have to be passed as a tuple.
Once the data is returned will add a column to the data frame named Select, and set its value to False. The column will shown as a checkbox in the data_editor widget.
Next we reset the column order so the first column is the Select column.
Lastly, return the streamlit data_editor widget to the user.
prompt_for_confirmation
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
def delete_source_data(self, delete_data: pd.DataFrame): """ After user confirms the deletion execute the delete commands against the source database Parameters ---------- edited_data : pandas DataFrame DataFrame only contains the rows to be deleted. """ self.deletion_requested = False # Retrieve selected rows from session state for deletion for row_id in delete_data[self.primary_key_column_name]: self.data_handler.delete_row(self._source_table_name, primary_key_column=self.primary_key_column_name, row_id=row_id) self._query_success = True |
First will reset the flag deletion_requested to False, so we don’t try and delete the same rows twice, if the deletion fails.
Next will extract all of the primary key ids that have been selected. In the delete_row.py callback to delete_source_data, we filtered the rows being passed to the method through the args.
|
1 |
args=(table_data.query("Select == True"),)) |
The Pandas data frame query method ensures we are only deleting rows that the user selected.
Next page will talk about the DataHandler Class
Leave a Reply