Le script python suivant se connecte a un compte de stockage Azure avec une cle de stockage et liste les proprietes des blobs d’un container, avec la possibilité d’exporter le resultat en csv.
Le parametre MaxResult est important et est directement lié au fait que la requete vers le container (voir fonction « list_all_blobs_in_container_by_lot ») s’effectue en generant un iterator (generator) qui recupere par lot (MaxResult) la liste des blobs.
Le parametre export_all_blobs_list est a False par defaut pour eviter de generer un fichier csv si le nombre de blobs est tres important (1 ligne/blob)
### azure_list_blobs_properties.py ###
##### Script Python pour lister les proprietes de blobs dans un container Azure Blob Storage #####
# PARAMETERES D'ENTREE DU SCRIPT :
# --storage_account_name : Nom du compte de stockage Azure (default: mystorageaccount)
# --Cledestockage : Clé de stockage Azure
# --nomcontainer : Nom du container de stockage Azure (default: mycontainer)
# --MaxResults : Nombre maximum de blobs à récupérer par lot (default: 10000)
# --export_all_blobs_list : Exporter la liste de tous les blobs dans un fichier CSV (True/False) (default: False)
# --exportfolder : Chemin du dossier pour les fichiers CSV (default: .)
# --fichierrapport : Chemin du fichier de rapport (default: myreport.txt)
# --help : Afficher l'aide du script
# USAGE :
# WARNING: Ne pas pas positionner a True le parametre export_all_blobs_list si le container cible contiens beaucoup de blobs. Le fichier csv generé sera très volumineux (1 ligne / blob).
# python.exe .\azure_list_blobs_properties.py --storage_account_name <nom_du_compte_de_stockage> --Cledestockage <clé_de_stockage> --nomcontainer <nom_du_container> --export_all_blobs_list True/False
import os
import argparse
from datetime import datetime, timezone
from azure.storage.blob import BlobServiceClient
import csv
if __name__ == '__main__':
###########################################
# Arguments du script
###########################################
parser = argparse.ArgumentParser()
parser.add_argument(
'--storage_account_name',
type=str,
default='mystorageaccount',
help='Nom du compte de stockage Azure'
)
parser.add_argument(
'--Cledestockage',
type=str,
help='Clé de stockage Azure'
)
parser.add_argument(
'--nomcontainer',
type=str,
default='mycontainer',
help='nom du container de stockage Azure'
)
parser.add_argument(
'--MaxResults',
type=int,
default=10000,
help='Nombre maximum de blobs à récupérer par lot (par défaut: 10000)'
)
parser.add_argument(
'--export_all_blobs_list',
choices=['True', 'False'],
default='False',
help='Exporter la liste de tous les blobs dans un fichier CSV'
)
# export folder
parser.add_argument(
'--exportfolder',
type=str,
default='.',
help='Chemin du dossier pour les fichiers CSV'
)
# fichier de rapport
parser.add_argument(
'--fichierrapport',
type=str,
default='myreport.txt',
help='Nom du fichier de rapport'
)
# On parse les arguments
args = parser.parse_args()
# AUTRES PARAMETRES
# URL du compte de stockage Azure
account_url = f"https://{args.storage_account_name}.blob.core.windows.net"
# On signifie a Python que l'on utilise PAS de proxy pour les connexions vers Azure
os.environ['NO_PROXY'] = '.blob.core.windows.net,login.microsoftonline.com,management.azure.com'
### FONCTION POUR ECRIRE UN MESSAGE DANS UN FICHIER DE RAPPORT
def write_to_report_file(message, filename):
"""
Écrit un message dans un fichier de rapport.
:param message: Message à écrire dans le fichier.
:param
filename: Nom du fichier de rapport.
"""
try:
with open(args.fichierrapport, 'a', encoding='utf-8') as fichier:
fichier.write(message + '\n')
print(f"Message ecrit dans le fichier de rapport '{args.fichierrapport}'.")
except Exception as e:
print(f"Erreur lors de l'ecriture dans le fichier {args.fichierrapport}: {e}")
# Fin de la fonction write_to_report_file
## FONCTION POUR TESTER L'ACCES AU CONTAINER DE STOCKAGE AZURE
def test_access_to_azure_storage_container(account_url, credential, container_name):
"""
Teste l'accès à un container de stockage Azure Blob.
:param account_url: URL du compte de stockage Azure.
:param credential: Clé d'accès ou SAS token pour le compte de stockage.
:param container_name: Nom du container à tester.
"""
try:
blob_service_client = BlobServiceClient(account_url=account_url, credential=credential)
container_client = blob_service_client.get_container_client(container_name)
# Vérification de la connexion
container_properties = container_client.get_container_properties()
print(f"Connexion reussie au container '{container_name}'.")
return True
except Exception as e:
print(f"Erreur lors de l'accès au container '{container_name}' : {e}")
return False
# Fin de la fonction test_access_to_azure_storage_container
## FONCTION POUR LISTER TOUT LES BLOBS DANS UN CONTAINER
def list_all_blobs_in_container_by_lot(container_client, max_results):
"""
Liste les blobs dans un conteneur Azure Blob Storage par lot avec une limite.
:param container_name: Nom du conteneur.
:param max_results: Nombre maximum de blobs à récupérer par lot.
:return: Un générateur qui retourne les blobs par lot.
"""
try:
# Liste les blobs par lot
blob_iterator = container_client.list_blobs(results_per_page=max_results).by_page()
for blob_page in blob_iterator:
yield list(blob_page) # Retourne un lot de blobs sous forme de liste
except Exception as e:
print(f"Erreur lors de la récupération des blobs : {e}")
# FONCTION POUR ECRIRE UN CSV
def write_csv(data, filename):
"""
Écrit les données dans un fichier CSV.
:param data: Liste de dictionnaires contenant les données à écrire.
Chaque dictionnaire représente une ligne, avec les clés comme en-têtes.
:param filename: Nom du fichier CSV à créer.
"""
try:
# Vérifier si la liste de données n'est pas vide
if not data:
print("Aucune donnee a ecrire dans le fichier CSV.")
return
# Ouvrir le fichier en mode écriture
with open(filename, mode='w', newline='', encoding='utf-8') as csvfile:
# Créer un objet writer
writer = csv.DictWriter(csvfile, fieldnames=data[0].keys())
# Écrire l'en-tête
writer.writeheader()
# Écrire les lignes
writer.writerows(data)
print(f"Les donnees ont ete ecrites avec succes dans le fichier '{filename}'.")
except Exception as e:
print(f"Erreur lors de l'ecriture dans le fichier CSV : {e}")
#Debut du traitement
print(f"Debut du traitement pour le container {args.nomcontainer}.")
# TEST DE L'ACCES AU CONTAINER DE STOCKAGE AZURE
print("Test de l'acces au container de stockage Azure...")
test_access_to_azure_storage_container(
account_url=account_url,
credential=args.Cledestockage,
container_name=args.nomcontainer
)
# Fin du test de l'accès au container de stockage Azure
# #################################################
# On recupere la date d'aujourd'hui
today = datetime.now().replace(tzinfo=timezone.utc)
# Instanciation du client de service Blob
container = BlobServiceClient(
account_url=account_url,
credential=args.Cledestockage
).get_container_client(args.nomcontainer)
all_blobs_generator = list_all_blobs_in_container_by_lot(
container_client=container,
max_results=args.MaxResults # Nombre maximum de blobs à récupérer par lot
)
# On verifie que le generator n'est pas vide
if not all_blobs_generator:
print(f"Aucun blob trouve dans le container '{args.nomcontainer}'.")
exit(1) # Sortie du script si aucun blob n'est trouvé
# Fin de la récupération des blobs dans le container
# Initialisation de la liste pour stocker les infos de tout les blobs (utilisé uniquement si l'option export_all_blobs_list est a True)
all_blobs_info = []
# On parcourt tous les blobs du container
print("\n"*2)
print("#"* 80)
print(f"Analyse et traitement des blobs dans le container {args.nomcontainer}...")
print("#"* 80)
# initialisation de l'index pour afficher le numero du lot et le nombre total de lot a la fin du traitement
lotcount = 0
# initialisation de l'index pour incrementer le nombre de blobs
blobcount = 0
try:
for blob_lot in all_blobs_generator:
lotcount += 1
print(f"Nombre de blobs dans le lot {lotcount}: {len(blob_lot)}")
for blob in blob_lot:
# On incremente le nombre de blobs
blobcount += 1
# On recupere le client du blob
blob_client = container.get_blob_client(blob.name)
# Récupération des propriétés du blob
blob_properties = blob_client.get_blob_properties()
# On alimente all_blobs_info
all_blobs_info.append({
'Container': blob.container,
'Blob_Name': blob.name,
'Blob_Size': blob.size,
'is_deleted': blob.deleted,
'remaining_retention_days': blob.remaining_retention_days,
'is_current_version': blob.is_current_version,
'last_modified': blob.last_modified,
'expiration': blob_properties.immutability_policy.expiry_time if blob_properties.immutability_policy else None
})
except StopIteration:
print("Fin de l'iterateur, tous les lots de blobs ont ete analyses et traites.")
except Exception as e:
print(f"Erreur lors de la recuperation des blobs : {e}")
# Fin de la récupération des blobs dans le container
# affichage du nombre lots analyses
print("\n")
print("#"* 80)
print(f"Nombre total de lots de blobs analyses: {lotcount}")
print("#"* 80)
# STATISTIQUES SUR LES BLOBS
print("\n")
print("#"* 80)
print(f"Nombre total de blobs dans le container {args.nomcontainer}: {blobcount}")
print("#"* 80)
# On affiche certaines proprietes de all_blobs_info
print("\n")
print("#"* 80)
print(f"Contenu de la liste de tous les blobs du container {args.nomcontainer}:")
for blob_info in all_blobs_info:
print(f"Blob Name: {blob_info['Blob_Name']}, Size: {blob_info['Blob_Size']} octets, Expiration: {blob_info['expiration']}, Last Modified: {blob_info['last_modified']}")
print("#"* 80)
# ECRIRE LE CONTENU DE all_blobs_info DANS UN CSV (Si l'option export_all_blobs_list est a True)
if args.export_all_blobs_list == 'True':
print("\n")
print("#"* 80)
print(f"Ecriture de la liste de tout les blobs du container dans le fichier CSV avec la date d'expiration...")
csv_filepath = f"{args.exportfolder}\\{args.nomcontainer}_all_blobs_info.csv"
write_csv(data=all_blobs_info, filename=csv_filepath)
print("#"* 80)
# On ajoute un message de fin dans le fichier de rapport avec toutes les statistiques
print("\n")
print("#"* 80)
msg = f"Fin du traitement pour le container {args.nomcontainer}\nLe traitement s'est opere en {lotcount} lots\nNombre total de blobs: {blobcount}\nFichier CSV des blobs: {csv_filepath if args.export_all_blobs_list == 'True' else 'Non exporté'}"
print(msg)
# Ecriture dans le fichier de rapport
write_to_report_file(
message = f"\n### {today.isoformat()} ###\n{msg}\n#########################################",
filename=f"{args.exportfolder}\\{args.fichierrapport}"
)
print("#"* 80)
print("\n")
# Fin du script
print("#"* 40 + " FIN DU SCRIPT " + "#"*40)
0 commentaires