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 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157
|
# coding: utf-8
# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
# --------------------------------------------------------------------------
"""
FILE: blob_samples_hello_world.py
DESCRIPTION:
This sample demos basic blob operations like getting a blob client from container, uploading and downloading
a blob using the blob_client.
USAGE: python blob_samples_hello_world.py
Set the environment variables with your own values before running the sample:
1) AZURE_STORAGE_CONNECTION_STRING - the connection string to your storage account
"""
import os
# set up
SOURCE_FILE = 'SampleSource.txt'
DEST_FILE = 'BlockDestination.txt'
class BlobSamples(object):
connection_string = os.getenv("AZURE_STORAGE_CONNECTION_STRING")
#--Begin Blob Samples-----------------------------------------------------------------
def create_container_sample(self):
# Instantiate a new BlobServiceClient using a connection string
from azure.storage.blob import BlobServiceClient
blob_service_client = BlobServiceClient.from_connection_string(self.connection_string)
# Instantiate a new ContainerClient
container_client = blob_service_client.get_container_client("mycontainer")
try:
# Create new container in the service
container_client.create_container()
# List containers in the storage account
list_response = blob_service_client.list_containers()
finally:
# Delete the container
container_client.delete_container()
def block_blob_sample(self):
# Instantiate a new BlobServiceClient using a connection string
from azure.storage.blob import BlobServiceClient
blob_service_client = BlobServiceClient.from_connection_string(self.connection_string)
# Instantiate a new ContainerClient
container_client = blob_service_client.get_container_client("myblockcontainersync")
try:
# Create new Container in the service
container_client.create_container()
# Instantiate a new BlobClient
blob_client = container_client.get_blob_client("myblockblob")
# [START upload_a_blob]
# Upload content to block blob
with open(SOURCE_FILE, "rb") as data:
blob_client.upload_blob(data, blob_type="BlockBlob")
# [END upload_a_blob]
# [START download_a_blob]
with open(DEST_FILE, "wb") as my_blob:
download_stream = blob_client.download_blob()
my_blob.write(download_stream.readall())
# [END download_a_blob]
# [START delete_blob]
blob_client.delete_blob()
# [END delete_blob]
finally:
# Delete the container
container_client.delete_container()
def page_blob_sample(self):
# Instantiate a new BlobServiceClient using a connection string
from azure.storage.blob import BlobServiceClient
blob_service_client = BlobServiceClient.from_connection_string(self.connection_string)
# Instantiate a new ContainerClient
container_client = blob_service_client.get_container_client("mypagecontainersync")
try:
# Create new Container in the Service
container_client.create_container()
# Instantiate a new BlobClient
blob_client = container_client.get_blob_client("mypageblob")
# Upload content to the Page Blob
data = b'abcd'*128
blob_client.upload_blob(data, blob_type="PageBlob")
# Download Page Blob
with open(DEST_FILE, "wb") as my_blob:
download_stream = blob_client.download_blob()
my_blob.write(download_stream.readall())
# Delete Page Blob
blob_client.delete_blob()
finally:
# Delete container
container_client.delete_container()
def append_blob_sample(self):
# Instantiate a new BlobServiceClient using a connection string
from azure.storage.blob import BlobServiceClient
blob_service_client = BlobServiceClient.from_connection_string(self.connection_string)
# Instantiate a new ContainerClient
container_client = blob_service_client.get_container_client("myappendcontainersync")
try:
# Create new Container in the Service
container_client.create_container()
# Instantiate a new BlobClient
blob_client = container_client.get_blob_client("myappendblob")
# Upload content to the Page Blob
with open(SOURCE_FILE, "rb") as data:
blob_client.upload_blob(data, blob_type="AppendBlob")
# Download Append Blob
with open(DEST_FILE, "wb") as my_blob:
download_stream = blob_client.download_blob()
my_blob.write(download_stream.readall())
# Delete Append Blob
blob_client.delete_blob()
finally:
# Delete container
container_client.delete_container()
if __name__ == '__main__':
sample = BlobSamples()
sample.create_container_sample()
sample.block_blob_sample()
sample.append_blob_sample()
sample.page_blob_sample()
|