File size: 5,327 Bytes
761fac5 |
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 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 |
import os
import streamlit as st
from azure.cosmos import CosmosClient, PartitionKey
from azure.storage.blob import BlobServiceClient
from azure.cosmos.exceptions import CosmosResourceNotFoundError
import requests
import glob
from datetime import datetime
# Initialize Azure Clients
COSMOS_CONNECTION_STRING = os.getenv('COSMOS_CONNECTION_STRING')
BLOB_STORAGE_CONNECTION_STRING = os.getenv('BLOB_STORAGE_CONNECTION_STRING')
cosmos_client = CosmosClient.from_connection_string(COSMOS_CONNECTION_STRING)
blob_service = BlobServiceClient.from_connection_string(BLOB_STORAGE_CONNECTION_STRING)
# Set page config
st.set_page_config(page_title="Azure Manager", page_icon="☁️", layout="wide")
# Function to Delete All Items in a Container
def delete_all_items_in_container(db_name, container_name):
database_client = cosmos_client.get_database_client(db_name)
container_client = database_client.get_container_client(container_name)
for item in container_client.read_all_items():
try:
partition_key = '/id'
container_client.delete_item(item=item['id'], partition_key=partition_key)
st.write(f"🗑️ Deleted Item: {item['id']}")
except CosmosResourceNotFoundError:
st.error(f"❌ Item not found: {item['id']}")
# Display and Manage Cosmos DB Structure
def display_and_manage_cosmos_db():
st.header('📊 Azure Cosmos DB Structure')
for db_properties in cosmos_client.list_databases():
db_name = db_properties['id']
st.subheader(f"🗄️ Database: {db_name}")
database_client = cosmos_client.get_database_client(db_name)
for container_properties in database_client.list_containers():
container_name = container_properties['id']
st.markdown(f"📁 **Container**: {container_name}")
container_client = database_client.get_container_client(container_name)
for item in container_client.read_all_items():
col1, col2, col3 = st.columns([3, 1, 1])
with col1:
st.markdown(f"📄 Item: `{item['id']}`")
if 'file_name' in item:
st.image(item['file_name'], caption=item['file_name'], width=200)
with col2:
if st.button(f"🗑️ Delete", key=f"delete_{item['id']}"):
partition_key = '/id'
container_client.delete_item(item=item['id'], partition_key=partition_key)
st.success(f"✅ Deleted Item: {item['id']}")
st.rerun()
# Insert PNG Images with Unique Identifiers
def insert_png_images_with_unique_ids(db_name, container_name):
container_client = cosmos_client.get_database_client(db_name).get_container_client(container_name)
png_files = glob.glob('*.png')
for file_name in png_files:
unique_id = f"{os.path.splitext(file_name)[0]}_{datetime.now().strftime('%Y%m%d%H%M%S')}"
item_data = {"id": unique_id, "file_name": file_name}
container_client.create_item(body=item_data)
st.write(f"📥 Inserted Item: {unique_id}")
# Streamlit UI
st.title("☁️ Azure Manager")
# Sidebar for global actions
st.sidebar.header("🛠️ Global Actions")
# Azure Blob Storage - Upload/Download
st.sidebar.subheader('📤 Azure Blob Storage - Upload')
blob_container = st.sidebar.text_input('🗂️ Blob Container')
blob_file = st.sidebar.file_uploader('📁 Upload file to Blob')
if blob_file is not None and st.sidebar.button('📤 Upload to Blob'):
blob_client = blob_service.get_blob_client(container=blob_container, blob=blob_file.name)
blob_client.upload_blob(blob_file.getvalue())
st.sidebar.success('✅ File uploaded successfully.')
# Azure Functions - Trigger
st.sidebar.subheader('⚡ Azure Functions - Trigger')
function_url = st.sidebar.text_input('🔗 Function URL')
if st.sidebar.button('🚀 Call Azure Function'):
response = requests.get(function_url)
st.sidebar.write('📡 Function Response:', response.text)
# Insert PNG Images
st.sidebar.subheader('🖼️ Insert PNG Images')
db_name_insert = st.sidebar.selectbox("🗄️ Select Database", [db['id'] for db in cosmos_client.list_databases()], key='db_insert')
container_name_insert = st.sidebar.selectbox("📁 Select Container", [container['id'] for container in cosmos_client.get_database_client(db_name_insert).list_containers()], key='container_insert')
if st.sidebar.button('📥 Insert PNG Images with Unique IDs'):
insert_png_images_with_unique_ids(db_name_insert, container_name_insert)
# Delete All Items in a Container
st.sidebar.subheader('🗑️ Delete All Items')
db_name_delete = st.sidebar.selectbox("🗄️ Select Database to Delete From", [db['id'] for db in cosmos_client.list_databases()], key='db_delete')
container_name_delete = st.sidebar.selectbox("📁 Select Container to Delete From", [container['id'] for container in cosmos_client.get_database_client(db_name_delete).list_containers()], key='container_delete')
if st.sidebar.button('🗑️ Delete All Items in Container'):
delete_all_items_in_container(db_name_delete, container_name_delete)
st.sidebar.success("✅ All items deleted successfully.")
st.rerun()
# Main content
display_and_manage_cosmos_db() |