pip install azure-storage-blob python-dotenv azure-core配置檔如下：
AZURE_CONNECTION_STRING=
AZURE_CONTAINER_NAME=
import argparse
import os
from typing import Optional
from azure.storage.blob import BlobServiceClient, ContainerClient, BlobClient
from dotenv import load_dotenv
# 載入.env
load_dotenv()
# 設定連接字串
AZURE_CONNECTION_STRING = os.getenv("AZURE_CONNECTION_STRING", None)
# 設定容器名稱
AZURE_CONTAINER_NAME = os.getenv("AZURE_CONTAINER_NAME", None)
# 如果不存在就拋出錯誤
if not AZURE_CONNECTION_STRING:
raise ValueError("Azure connection string is not provided.")
# 全域變數 BlobServiceClient
azure_blob_service_client = BlobServiceClient.from_connection_string(AZURE_CONNECTION_STRING)
def get_container_client(container_name: str = AZURE_CONTAINER_NAME) -> Optional[ContainerClient]:
"""
取得指定容器的 ContainerClient，如果容器不存在則建立新容器
Args:
container_name: 容器名稱，如果未指定則使用環境變數中的設定
Returns:
ContainerClient: 成功返回容器客戶端對象
None: 如果發生錯誤
"""
# 取得container client
client = azure_blob_service_client.get_container_client(AZURE_CONTAINER_NAME.lower())
# 如果存在就回傳
if client.exists():
return client
else:
# 如果不存在就建立
return create_container(container_name)
def create_container(container_name: str = None) -> Optional[ContainerClient]:
"""
建立新的 Blob 存儲容器，如果容器已存在則返回現有容器
Args:
container_name: 要建立的容器名稱
Returns:
ContainerClient: 成功返回容器客戶端對象
None: 如果發生錯誤
Raises:
ValueError: 未提供容器名稱時拋出
"""
global AZURE_CONTAINER_NAME
# 如果名稱不存在就拋出錯誤
if not container_name:
raise ValueError("Container name is not provided.")
try:
# 將容器名稱轉為小寫
container_name_lower = container_name.lower()
# 直接檢查容器是否存在
container_client = azure_blob_service_client.get_container_client(container_name_lower)
if container_client.exists():
return container_client
# 不存在則建立新容器
AZURE_CONTAINER_NAME = container_name_lower
print(f"Container '{AZURE_CONTAINER_NAME}' created successfully.")
return azure_blob_service_client.create_container(AZURE_CONTAINER_NAME)
except Exception as e:
print(f"Error creating container: {e}")
return None
def delete_container(container_name: str = AZURE_CONTAINER_NAME) -> bool:
"""
刪除指定的 Blob 存儲容器
Args:
container_name: 要刪除的容器名稱，如果未指定則使用環境變數中的設定
Returns:
bool: 刪除成功返回 True，失敗返回 False
Raises:
ValueError: 未提供容器名稱時拋出
"""
try:
# 如果容器名稱不存在就拋出錯誤
if not container_name:
raise ValueError("Container name is not provided")
# 直接獲取容器客戶端，而不是使用 get_container_client()
container_client = azure_blob_service_client.get_container_client(container_name.lower())
# 如果容器不存在就回傳
if not container_client.exists():
print(f"Container '{container_name}' does not exist")
return False
# 刪除容器
container_client.delete_container()
print(f"Container '{container_name}' deleted successfully")
return True
except Exception as e:
print(f"Error deleting container: {e}")
return False
def get_blob_client(blob_name: str) -> BlobClient:
"""
取得指定 Blob 的客戶端對象
Args:
blob_name: Blob 的名稱
Returns:
BlobClient: Blob 客戶端對象
Raises:
ValueError: 容器客戶端不可用時拋出
"""
container_client = get_container_client()
if not container_client:
raise ValueError("Container client not available")
return container_client.get_blob_client(blob_name)
def upload_blob(file_path: str) -> None:
"""
上傳檔案到 Blob 存儲
Args:
file_path: 要上傳的檔案路徑（支持相對和絕對路徑）
Returns:
None
Raises:
FileNotFoundError: 檔案不存在時拋出
ValueError: 容器客戶端不可用時拋出
"""
try:
# 開啟檔案
with open(file_path, "rb") as data:
# 取得檔案名稱
blob_name = os.path.basename(file_path)
# 取得blob client
blob_client = get_blob_client(blob_name)
# 上傳blob 並且覆蓋相同檔案名稱的檔案
blob_client.upload_blob(data, overwrite=True)
print(f"Blob '{blob_name}' uploaded successfully.")
except Exception as e:
print(f"Error uploading blob: {e}")
def download_blob(blob_name: str, download_path: str) -> bool:
"""
從 Blob 存儲下載檔案
Args:
blob_name: 要下載的 Blob 名稱
download_path: 下載檔案的保存路徑（會自動創建目錄）
Returns:
bool: 下載成功返回 True，失敗返回 False
Raises:
ValueError: 容器客戶端不可用時拋出
"""
try:
blob_client = get_blob_client(blob_name)
# 確保下載目錄存在
os.makedirs(os.path.dirname(os.path.abspath(download_path)), exist_ok=True)
# 下載檔案
with open(download_path, "wb") as download_file:
download_file.write(blob_client.download_blob().readall())
print(f"Blob '{blob_name}' 下載成功，保存至 '{download_path}'")
return True
except Exception as e:
print(f"下載 blob 時發生錯誤: {e}")
return False
def delete_blob(blob_name: str) -> bool:
"""
刪除指定的 Blob
Args:
blob_name: 要刪除的 Blob 名稱
Returns:
bool: 刪除成功返回 True，失敗返回 False
Raises:
ValueError: 容器客戶端不可用時拋出
"""
try:
# 取得blob client
blob_client = get_container_client().get_blob_client(blob_name)
# 如果存在就刪除
if blob_client.exists():
# 刪除blob
blob_client.delete_blob()
print(f"Blob '{blob_name}' deleted successfully.")
return True
else:
print(f"Blob '{blob_name}' does not exist.")
return False
except Exception as e:
print(f"Error deleting blob: {e}")
return False
def list_blobs() -> None:
"""
列出容器中的所有 Blobs
打印格式：
Found X blob(s):
- blob1_name
- blob2_name
...
Returns:
None
Raises:
ValueError: 容器客戶端不可用時拋出
"""
try:
# 將 blobs 轉換成列表
blob_list = list(get_container_client().list_blobs())
# 檢查是否有 blobs
if blob_list:
print(f"Found {len(blob_list)} blob(s):")
for blob in blob_list:
print(f"- {blob.name}")
else:
print("No blobs found in the container.")
except Exception as e:
print(f"Error listing blobs: {e}")
def main():
"""
主程式入口，處理命令行參數並執行相應操作
支持的操作：
- upload: 上傳檔案
- download: 下載檔案
- delete: 刪除 Blob
- list: 列出所有 Blobs
- create_container: 創建容器
- delete_container: 刪除容器
Returns:
int: 0 表示成功，1 表示失敗
"""
parser = argparse.ArgumentParser(
description='Azure Blob Storage 操作工具',
formatter_class=argparse.RawTextHelpFormatter
)
# 定義操作類型，添加 download 選項
actions_help = """操作類型:
upload 上傳文件到 Blob 存儲
download 從 Blob 存儲下載文件
delete 刪除指定的 Blob
list 列出容器中的所有 Blobs
create_container 創建新的存儲容器
delete_container 刪除指定的存儲容器"""
parser.add_argument(
'action',
choices=['upload', 'download', 'delete', 'list', 'create_container', 'delete_container'],
help=actions_help
)
parser.add_argument(
'--file',
metavar='FILE_PATH',
help='文件路徑（上傳時為源文件路徑，下載時為保存路徑）'
)
parser.add_argument(
'--blob',
metavar='BLOB_NAME',
help='Blob 名稱（用於下載、刪除操作）'
)
parser.add_argument(
'--container',
metavar='CONTAINER_NAME',
help='容器名稱（配合 create_container 或 delete_container 使用）'
)
args = parser.parse_args()
try:
if args.action == 'upload':
if not args.file:
raise ValueError("上傳操作需要提供文件路徑 (--file)")
upload_blob(args.file)
elif args.action == 'download':
if not args.blob or not args.file:
raise ValueError("下載操作需要提供 Blob 名稱 (--blob) 和保存路徑 (--file)")
download_blob(args.blob, args.file)
elif args.action == 'delete':
if not args.blob:
raise ValueError("刪除操作需要提供 Blob 名稱 (--blob)")
delete_blob(args.blob)
elif args.action == 'list':
list_blobs()
elif args.action == 'create_container':
if not args.container:
raise ValueError("創建容器操作需要提供容器名稱 (--container)")
create_container(args.container)
elif args.action == 'delete_container':
container_name = args.container if args.container else AZURE_CONTAINER_NAME
delete_container(container_name)
except Exception as e:
print(f"Error: {e}")
return 1
return 0
if __name__ == '__main__':
exit(main())