access_az_blob.py

import io
import logging
import os

from azure.storage.blob import BlobClient, BlobServiceClient, ContainerClient

BLOB_CONNECT_STR = ""
conn_str = os.environ.get(
    "AzureWebJobsStorage",
    BLOB_CONNECT_STR    )
blob_service_client = BlobServiceClient.from_connection_string(conn_str)

CONTAINER_NAME = ""


def upload_file(byte_io, remote_path, container_name=None):
    if container_name is not None:
        blob_client = blob_service_client.get_blob_client(
            container=container_name, blob=remote_path
        )
    else:
        container_client = blob_service_client.get_container_client(CONTAINER_NAME)
        blob_client = container_client.get_blob_client(blob=remote_path)

    try:
        blob_client.upload_blob(byte_io, overwrite=True)
    except Exception:
        logging.warning(
            f"upload {blob_client.container_name}/{remote_path} failed "
        )


def download_file(remote_path, container_name=None) -> io.BytesIO:
    if container_name is not None:
        blob_client = blob_service_client.get_blob_client(
            container=container_name, blob=remote_path
        )
    else:
        container_client = blob_service_client.get_container_client(CONTAINER_NAME)
        blob_client = container_client.get_blob_client(blob=remote_path)

    try:
        return blob_client.download_blob().readall()
    except Exception:
        logging.warning(
            f"download {blob_client.container_name}/{remote_path} failed "
        )

    return None


file_io = download_file('file.txt')

if file_io is not None:
#     str_val = file_io.decode('utf-8')
#     with open('./sample.yaml', 'w') as f:
#         f.write(str_val)

    upload_file(file_io, 'file_copy.txt', container_name='testing')