87 lines
3.5 KiB
Python
87 lines
3.5 KiB
Python
import requests
|
|
import json
|
|
|
|
from azure.storage.blob import BlobServiceClient, ContentSettings
|
|
from azure.storage.queue import QueueClient
|
|
from azure.identity import DefaultAzureCredential
|
|
from azure.keyvault.secrets import SecretClient
|
|
from azure.identity import DefaultAzureCredential
|
|
|
|
KEY_VAULT_NAME = "ado2-keys"
|
|
def get_keyvault_client():
|
|
key_vault_uri = f"https://{KEY_VAULT_NAME}.vault.azure.net/"
|
|
|
|
# DefaultAzureCredential이 Managed Identity를 자동으로 사용
|
|
credential = DefaultAzureCredential()
|
|
return SecretClient(vault_url=key_vault_uri, credential=credential)
|
|
|
|
# Secret Key
|
|
BLOB_ACCOUNT_URL_KEY = "ado2-blob-account-url"# https://mystorageaccount.blob.core.windows.net
|
|
QUEUE_URL_KEY = "ado2-queue-url"
|
|
|
|
# Plain text
|
|
BLOB_CONTAINER_NAME = "ado2-media-public-access"
|
|
BLOB_MEDIA_FOLDER = "ado2-media-original"
|
|
|
|
def az_storage_upload_ado2_media(data, remote_file_path:str):
|
|
secret_client = get_keyvault_client()
|
|
account_url = secret_client.get_secret(BLOB_ACCOUNT_URL_KEY).value
|
|
# Managed Identity 사용
|
|
credential = DefaultAzureCredential()
|
|
blob_service_client = BlobServiceClient(account_url=account_url, credential=credential)
|
|
|
|
blob_client = blob_service_client.get_blob_client(container=BLOB_CONTAINER_NAME, blob=f"{BLOB_MEDIA_FOLDER}/{remote_file_path}")
|
|
|
|
# with open(local_file_path, 'rb') as data:
|
|
# blob_client.upload_blob(data, overwrite=True)
|
|
blob_client.upload_blob(data, overwrite=True)
|
|
return blob_client.url
|
|
|
|
|
|
def az_storage_read_ado2_media(remote_file_path:str):
|
|
secret_client = get_keyvault_client()
|
|
account_url = secret_client.get_secret(BLOB_ACCOUNT_URL_KEY).value
|
|
# Managed Identity 사용
|
|
credential = DefaultAzureCredential()
|
|
blob_service_client = BlobServiceClient(account_url=account_url, credential=credential)
|
|
|
|
blob_client = blob_service_client.get_blob_client(container=BLOB_CONTAINER_NAME, blob=f"{BLOB_MEDIA_FOLDER}/{remote_file_path}")
|
|
|
|
blob_data = blob_client.download_blob()
|
|
return blob_data
|
|
|
|
def az_storage_get_ado2_media_list(remote_folder_path:str) -> list[str]:
|
|
secret_client = get_keyvault_client()
|
|
account_url = secret_client.get_secret(BLOB_ACCOUNT_URL_KEY).value
|
|
# Managed Identity 사용
|
|
credential = DefaultAzureCredential()
|
|
blob_service_client = BlobServiceClient(account_url=account_url, credential=credential)
|
|
|
|
prefix = f"{BLOB_MEDIA_FOLDER}/{remote_folder_path}"
|
|
blob_media_list = blob_service_client.get_container_client(BLOB_CONTAINER_NAME).list_blobs(name_starts_with=prefix)
|
|
return blob_media_list
|
|
|
|
def az_add_queue_message(body:dict):
|
|
secret_client = get_keyvault_client()
|
|
queue_url = secret_client.get_secret(QUEUE_URL_KEY).value
|
|
credential = DefaultAzureCredential()
|
|
queue_client = QueueClient.from_queue_url(queue_url, credential=credential)
|
|
queue_client.send_message(json.dumps(body, ensure_ascii=False))
|
|
|
|
def az_get_queue_message():
|
|
secret_client = get_keyvault_client()
|
|
queue_url = secret_client.get_secret(QUEUE_URL_KEY).value
|
|
credential = DefaultAzureCredential()
|
|
queue_client = QueueClient.from_queue_url(queue_url, credential=credential)
|
|
|
|
# 메시지 읽기
|
|
messages = queue_client.receive_messages()
|
|
data_list = []
|
|
for message in messages:
|
|
# 메시지 처리
|
|
data = json.loads(message.content)
|
|
print(f"받은 메시지: {data}")
|
|
# 처리 후 삭제
|
|
queue_client.delete_message(message.id, message.pop_receipt)
|
|
data_list.append(data)
|
|
return data_list |