import streamlit as st import pandas as pd import re from azure.identity import DefaultAzureCredential from azure.mgmt.resource import SubscriptionClient from azure.mgmt.compute import ComputeManagementClient from jinja2 import Template, Environment, FileSystemLoader import json from os import getenv from dotenv import load_dotenv load_dotenv() def is_valid(key_name: str): return key_name in st.session_state and st.session_state[key_name] is not None and len(st.session_state[key_name]) > 0 def is_valid_and_not_equal(key_name: str, value: str): return key_name in st.session_state and st.session_state[key_name] != value def clear_offers(): if 'offers' in st.session_state: del st.session_state.offers if 'selected_offer' in st.session_state: del st.session_state.selected_offer clear_skus() def clear_skus(): if 'skus' in st.session_state: del st.session_state.skus if 'selected_sku' in st.session_state: del st.session_state.selected_sku clear_image_versions() def clear_image_versions(): if 'image_versions' in st.session_state: del st.session_state.image_versions if 'selected_image_version' in st.session_state: del st.session_state.selected_image_version def on_publisher_changed(): clear_offers() # st.rerun() def on_offer_changed(): clear_skus() # st.rerun() def on_sku_changed(): clear_skus() # st.rerun() def on_image_version_changed(): clear_image_versions() # st.rerun() def version_key(v): return [int(x) for x in v.split('.')] @st.cache_data def get_publishers(location: str): return [publisher.name for publisher in compute_client.virtual_machine_images.list_publishers(location)] @st.cache_data def get_offers(location: str, publisher: str): return [offer.name for offer in compute_client.virtual_machine_images.list_offers(location, publisher)] @st.cache_data def get_skus(location: str, publisher: str, offer: str): return [sku.name for sku in compute_client.virtual_machine_images.list_skus(location, publisher, offer)] @st.cache_data def get_image_versions(location: str, publisher: str, offer: str, sku: str): return [version.name for version in compute_client.virtual_machine_images.list(location, publisher, offer, sku)] @st.cache_data def get_locations(): subscription_client = SubscriptionClient(credential) locations = sorted(subscription_client.subscriptions.list_locations(getenv('AZURE_SUBSCRIPTION_ID')), key=lambda l: l.name) return [ loc for loc in locations if loc.metadata.region_type == 'Physical' ] subscription_id = getenv("AZURE_SUBSCRIPTION_ID") default_location = getenv("AZURE_LOCATION") credential = DefaultAzureCredential() locations = get_locations() compute_client = ComputeManagementClient(credential, subscription_id) st.set_page_config(page_title='Azure Image Chooser', layout='wide') st.title('Azure Image Chooser') location_cols = st.columns(4) location = location_cols[0].selectbox('Select Azure Location', options=[{"name": loc.name, "display_name": loc.display_name} for loc in locations], index=locations.index(next((l for l in locations if l.name == default_location), default_location)), format_func=lambda loc: loc['display_name'] )['name'] publisher_col, offer_col, sku_col, version_col = st.columns(4) # Publishers if 'publishers' not in st.session_state: st.session_state.publishers = get_publishers(location) clear_offers() if is_valid('publishers'): st.session_state.selected_publisher = publisher_col.selectbox('Select Publisher', options=st.session_state.publishers, on_change=on_publisher_changed, index=None) else: st.error("No publishers found. Please check your Azure subscription and location.") st.stop() # Offers if 'offers' not in st.session_state and is_valid('selected_publisher'): st.session_state.offers = get_offers(location, st.session_state.selected_publisher) if is_valid('offers'): st.session_state.selected_offer = offer_col.selectbox('Select Offer', options=st.session_state.offers, on_change=on_offer_changed, index=None) elif is_valid('selected_publisher'): st.info("No offers found for the selected publisher. Please select a different publisher.") st.stop() else: st.stop() # SKUs if 'skus' not in st.session_state and is_valid('selected_publisher') and is_valid('selected_offer'): st.session_state.skus = get_skus(location, st.session_state.selected_publisher, st.session_state.selected_offer) if is_valid('skus'): st.session_state.selected_sku = sku_col.selectbox('Select SKU', options=st.session_state.skus, on_change=on_sku_changed, index=None) elif is_valid('selected_offer'): st.info("No SKUs found for the selected offer. Please select a different offer.") st.stop() else: st.stop() # Image versions if 'image_versions' not in st.session_state and is_valid('selected_publisher') and is_valid('selected_offer') and is_valid('selected_sku'): image_versions = get_image_versions(location, st.session_state.selected_publisher, st.session_state.selected_offer, st.session_state.selected_sku) # Check if all image version string match the re. regex = re.compile(r'^[0-9]+\.[0-9]+\.[0-9]+$') if all(regex.match(version) for version in image_versions): image_versions = sorted(image_versions, key=version_key) st.session_state.image_versions = image_versions if is_valid('image_versions'): st.session_state.selected_image_version = version_col.selectbox('Select Image Version', options=st.session_state.image_versions, index=None) elif is_valid('selected_sku'): st.info("No image versions found for the selected SKU. Please select a different SKU.") st.stop() else: st.stop() if is_valid('selected_image_version'): st.subheader("Usage example") with open("templates.json") as f: templates = json.load(f) def usage_scenario_label(item): return item['label'] layout = st.columns(4) selected_file = layout[0].selectbox('Select usage scenario:', options=templates, format_func=usage_scenario_label) env = Environment(loader=FileSystemLoader("templates")) tpl = env.get_template(selected_file['file']) rendered = tpl.render( publisher=st.session_state.selected_publisher, offer=st.session_state.selected_offer, sku=st.session_state.selected_sku, version=st.session_state.selected_image_version ) st.code(rendered, language=selected_file['language'])