Within Azure, file shares (fileshares) are one of the two chief systems of cloud storage, the other being blob storage. This guide will walk through an example of how the Python SDK can be used to manage your files in an Azure file share. We will walk through a Python program, covering authentication and interacting with the file share: listing items in a directory, uploading files, downloading files, and deleting files.
Prerequisites
If you’re following along with this guide, there are some prerequisites to be mentioned. You will need an account with Azure, and you will need to create a file share, which requires a storage account, which in turn requires a resource group. This is the file share we will interact with.
You also need to install azure-storage-file-share with pip.
Authentication
To provide our program with access to our storage account (which contains our file share), we will use its connection string. This is analogous to a private key in that you should protect it.
Navigate to your file share and the tab “Access keys”. Click “Show” by the connection string under key1.
We need to add an environment variable to our system containing that connection string. The variable name will be “AZURE_STORAGE_CONNECTION_STRING” and the variable value will be the full connection string. If you are unfamiliar with how to add an environment variable, see here.
We do this in an environment variable as a security measure. If we included the string directly in our code, and someone else somehow got our code, then they would be able to access our storage account.
Note that environment variable changes may not take effect in an ongoing program or terminal. If you are having issues, close and reopen it.
The Program
This program will parse various arguments. We will have options to list, download, upload, and delete files. We will also have directory and fileshare arguments, which will provide information for listing, downloading, uploading, and deleting.
We will get the connection string into the program as it runs by using os.getenv(“AZURE_STORAGE_CONNECTION_STRING”).
import os, argparse
from azure.storage.fileshare import ShareDirectoryClient, ShareFileClient
from azure.core.exceptions import ResourceNotFoundError
def main():
# Parse input
parser = argparse.ArgumentParser(description="Generate your vaccination QR code.")
parser.add_argument("-l", "--list", action='store_true', help="List elements in a directory.", required=False)
parser.add_argument("-d", "--directory", type=str, help="Directory to upload to, download from, or list.", required=False)
parser.add_argument("-f", "--fileshare", type=str, help="Fileshare to interact with.", required=False)
parser.add_argument("-dl", "--download", type=str, nargs=2, help="File to download, and where to put it.", required=False)
parser.add_argument("-ul", "--upload", type=str, help="File to upload.", required=False)
parser.add_argument("-x", "--delete", type=str, help="File to delete remotely.", required=False)
args = parser.parse_args()
# Get connection string from environment vairables, so we can connect to our storage acount.
connection_string = os.getenv("AZURE_STORAGE_CONNECTION_STRING")
First, let’s list all the items in a given directory. We will require a user-provided directory (or if none is given we will assume the top-level directory), and a fileshare name. We need a fileshare name because you can have many fileshares within the same storage account.
ShareDirectoryClient is a client giving us access to a particular directory. Once we have it (if the requested directory is invalid we will get an error) we can operate on it similar to a directory on your local computer.
# List items in the given directory, or if none is given, the top-level directory.
if args.list:
if not args.fileshare:
print("Provide a fileshare to be viewed.")
return
try:
parent_dir = ShareDirectoryClient.from_connection_string(conn_str=connection_string, share_name=args.fileshare, directory_path=args.directory)
mylist = list(parent_dir.list_directories_and_files())
for item in mylist:
print(item["name"])
except ResourceNotFoundError:
print(f"Could not find/open directory: {args.directory}")
Next, we’ll upload a file. We need a few arguments now: the fileshare, the file to be uploaded, and the target directory in cloud storage. If no target directory is provided then we assume the top-level directory. This function uses a ShareFileClient, which is similar to ShareDirectoryClient except now we are dealing with a particular file. In this case, we are identifying the location where we want our file to go, and the client will facilitate its transfer.
# Upload a given file to a given directory. If no directory is given, upload to the top-level directory.
if args.upload:
if not args.fileshare:
print("Provide a fileshare to be uploaded to.")
return
try:
targetPath = os.path.basename(args.upload)
if args.directory:
targetPath = os.path.join(args.directory, os.path.basename(args.upload))
file_client = ShareFileClient.from_connection_string(conn_str=connection_string, share_name=args.fileshare, file_path=targetPath)
with open(args.upload, "rb") as source_file:
file_client.upload_file(source_file)
except ResourceNotFoundError:
print(f"Upload failed: Directory {args.directory} does not exist.")
Downloading a file is similar. To download, we will need a fileshare, a path to a file that exists on the cloud, and a target location for it on our local computer. I have the –download argument set up to accept 2 values for this purpose: the source and the destination.
# Download a given target file to a given local location.
if args.download:
if not args.download[0]:
print("Provide a file to be downloaded (full path, including any directories).")
return
if not args.download[1]:
print("Provide a target filename (including path) where the file will be downloaded to.")
return
if not args.fileshare:
print("Provide a fileshare to be downloaded from.")
return
try:
file_client = ShareFileClient.from_connection_string(conn_str=connection_string, share_name=args.fileshare, file_path=args.download[0])
with open(args.download[1], "wb") as file_handle:
data = file_client.download_file()
data.readinto(file_handle)
except ResourceNotFoundError:
print(f"Download failed: File {args.download[0]} does not exist.")
Finally, we will create the functionality for deleting a file from cloud storage. We will access the file similar to above. Just like ShareDirectoryClient gives direct access to a directory, ShareFileClient gives direct access to a file, and we can then manipulate it just like we might on a local computer. We can simply delete it by calling delete_file().
# Delete a given target file from cloud storage.
if args.delete:
if not args.fileshare:
print("Provide a fileshare to be deleted from.")
return
try:
file = ShareFileClient.from_connection_string(conn_str=connection_string, share_name=args.fileshare, file_path=args.delete)
file.delete_file()
except ResourceNotFoundError:
print(f"Deletion failed: File {args.delete} does not exist.")
I personally like to have a called main function to structure tools like this.
main()
Here is the full program:
import os, argparse
from azure.storage.fileshare import ShareDirectoryClient, ShareFileClient
from azure.core.exceptions import ResourceNotFoundError
def main():
# Parse input
parser = argparse.ArgumentParser(description="Generate your vaccination QR code.")
parser.add_argument("-l", "--list", action='store_true', help="List elements in a directory.", required=False)
parser.add_argument("-d", "--directory", type=str, help="Directory to upload to, download from, or list.", required=False)
parser.add_argument("-f", "--fileshare", type=str, help="Fileshare to interact with.", required=False)
parser.add_argument("-dl", "--download", type=str, nargs=2, help="File to download, and where to put it.", required=False)
parser.add_argument("-ul", "--upload", type=str, help="File to upload.", required=False)
parser.add_argument("-x", "--delete", type=str, help="File to delete remotely.", required=False)
args = parser.parse_args()
# Get connection string from environment vairables, so we can connect to our storage acount.
connection_string = os.getenv("AZURE_STORAGE_CONNECTION_STRING")
# List items in the given directory, or if none is given, the top-level directory.
if args.list:
if not args.fileshare:
print("Provide a fileshare to be viewed.")
return
try:
parent_dir = ShareDirectoryClient.from_connection_string(conn_str=connection_string, share_name=args.fileshare, directory_path=args.directory)
mylist = list(parent_dir.list_directories_and_files())
for item in mylist:
print(item["name"])
except ResourceNotFoundError:
print(f"Could not find/open directory: {args.directory}")
# Upload a given file to a given directory. If no directory is given, upload to the top-level directory.
if args.upload:
if not args.fileshare:
print("Provide a fileshare to be uploaded to.")
return
try:
targetPath = os.path.basename(args.upload)
if args.directory:
targetPath = os.path.join(args.directory, os.path.basename(args.upload))
file_client = ShareFileClient.from_connection_string(conn_str=connection_string, share_name=args.fileshare, file_path=targetPath)
with open(args.upload, "rb") as source_file:
file_client.upload_file(source_file)
except ResourceNotFoundError:
print(f"Upload failed: Directory {args.directory} does not exist.")
# Download a given target file to a given local location.
if args.download:
if not args.download[0]:
print("Provide a file to be downloaded (full path, including any directories).")
return
if not args.download[1]:
print("Provide a target filename (including path) where the file will be downloaded to.")
return
if not args.fileshare:
print("Provide a fileshare to be downloaded from.")
return
try:
file_client = ShareFileClient.from_connection_string(conn_str=connection_string, share_name=args.fileshare, file_path=args.download[0])
with open(args.download[1], "wb") as file_handle:
data = file_client.download_file()
data.readinto(file_handle)
except ResourceNotFoundError:
print(f"Download failed: File {args.download[0]} does not exist.")
# Delete a given target file from cloud storage.
if args.delete:
if not args.fileshare:
print("Provide a fileshare to be deleted from.")
return
try:
file = ShareFileClient.from_connection_string(conn_str=connection_string, share_name=args.fileshare, file_path=args.delete)
file.delete_file()
except ResourceNotFoundError:
print(f"Deletion failed: File {args.delete} does not exist.")
main()
Now we can easily manipulate our storage account from the command line!
There is no end to what can be done within Azure via the Python SDK. This was a simple program, but you can also make powerful industry tools and customized automation, for example.