Python: How to generate string of arbitrary length of any alphabet characters
I was testing a bug where a field was limited to 255 characters only. I needed…
February 03, 2021
In this tutorial we will see:
And, everything will be in Python
This tutorial is based upon Python-3.7
We would require azure-storage-blob
. Code is tested with version 12.7.1
from azure.storage.blob import BlobServiceClient
# consider a dictionary container
container = {
'account_name': 'your_account_name',
'container_name': 'your_container_name',
'sas_token': 'xxxxxxx'
}
if "account_key" in container:
blob_service = BlobServiceClient(
account_url=account_url, credential=container["account_key"])
elif "sas_token" in container:
blob_service = BlobServiceClient(
account_url=account_url, credential=container["sas_token"])
else:
blob_service = BlobServiceClient(account_url=account_url)
# Now to get instance of class which has list_blob methods
container_client = blob_service.get_container_client(container['container_name'])
In above code, we are just instantiating client classes required for the operation and authenticate.
In my example, I have a sas_token
.
import os
from azure.storage.blob import BlobServiceClient
def _create_dirs(dest_path):
if not os.path.exists(dest_path):
os.makedirs(dest_path)
elif not os.path.isdir(dest_path):
shutil.rmtree(dest_path)
os.makedirs(dest_path)
def _get_container_service(container):
account_url = f'https://{container["account_name"]}.blob.core.windows.net'
proxies = None
if 'proxy' in container:
proxies = {'http': container['proxy']}
# If 'proxy' isn't specified in container block, check if 'https_proxy' is set.
elif 'https_proxy' in container:
proxies = {'https': container['https_proxy']}
# instantiate based upon credential
if "account_key" in container:
blob_service = BlobServiceClient(
account_url=account_url, credential=container["account_key"], proxies=proxies)
elif "sas_token" in container:
blob_service = BlobServiceClient(
account_url=account_url, credential=container["sas_token"], proxies=proxies)
else:
blob_service = BlobServiceClient(account_url=account_url, proxies=proxies)
return blob_service.get_container_client(container['container_name'])
def download_blobs(container, dest_path):
## You might want to handle some exceptions here
_create_dirs(dest_path)
# Get the container instance
blob_service = _get_container_service(container)
# Note: list_blobs returns an iterator
blob_list = blob_service.list_blobs()
for blob in blob_list:
fname = os.path.join(dest_path, blob.name)
print(f'Downloading {blob.name} to {fname}')
# get blob client which has download_blob method
blob_client = blob_service.get_blob_client(blob)
# create base dirs if not exists
_create_dirs(os.path.dirname(fname))
with open(fname, "wb") as download_file:
download_file.write(blob_client.download_blob().readall())
## main starts here
local_dest_path = './container_blob'
container = {
'account_name': 'your_account_name',
'container_name': 'your_container_name',
'sas_token': 'xxxxxxx'
}
download_blobs(container, local_dest_path)
Above script is very simple to understand. My container has nested directories and files. The code iterate over all files and downloads one by one.
{
'name': 'fdg/cert_discovery.fdg',
'snapshot': None,
'content': None,
'properties': {
'blob_type': 'BlockBlob',
'last_modified': datetime.datetime(2019, 12, 2, 9, 42, 50, tzinfo=tzutc()),
'etag': '0x8D7770BFF1CC8A1',
'content_length': 423,
'content_range': None,
'append_blob_committed_block_count': None,
'page_blob_sequence_number': None,
'server_encrypted': True,
'copy': {
'id': None,
'source': None,
'status': None,
'progress': None,
'completion_time': None,
'status_description': None
},
'content_settings': {
'content_type': 'application/octet-stream',
'content_encoding': None,
'content_language': None,
'content_disposition': None,
'cache_control': None,
'content_md5': '3ycLC3CutKkybJtlgvEdsQ=='
},
'lease': {
'status': 'unlocked',
'state': 'available',
'duration': None
},
'blob_tier': None,
'blob_tier_change_time': None,
'blob_tier_inferred': False,
'deleted_time': None,
'remaining_retention_days': None,
'creation_time': datetime.datetime(2019, 11, 28, 11, 52, 5, tzinfo=tzutc())
},
'metadata': None,
'deleted': False
}
For usage without Azure libraries, see: List and Download Azure blobs by Python Libraries
Let me know if you face any difficulties, and I will try to resolve them.
I was testing a bug where a field was limited to 255 characters only. I needed…
Tag the image, by seeing its image id, from docker images command docker tag 04d…
This will take backup of your passed database name, to the passed folder. It…
So, here we are using input variable String[] args without any validation…
While doing code review of one of junior, I encountered following code: What…
Listing down the commonly used Elastic Search queries. You can get search…
Introduction This post has the complete code to send email through smtp server…
Introduction In a normal email sending code from python, I’m getting following…
Introduction In one of my app, I was using to talk to . I have used some event…
Introduction So you have a Django project, and want to run it using docker image…
Introduction It is very important to introduce few process so that your code and…
Introduction In this post, we will see a sample Jenkin Pipeline Groovy script…