validate endpoint url

pyproject.toml

@@ -11,6 +11,7 @@ snakemake-interface-common = "^1.14.2"
 snakemake-interface-storage-plugins = "^1.2.3"
 azure-storage-blob = "^12.19.0"
 azure-core = "^1.29.5"
+azure-identity = "^1.15.0"
 
 
 [tool.poetry.group.dev.dependencies]

snakemake_storage_plugin_az/__init__.py

@@ -1,4 +1,5 @@
 from dataclasses import dataclass, field
+import re
 from azure.storage.blob import BlobServiceClient
 from typing import Any, Iterable, Optional
 from snakemake_interface_storage_plugins.settings import StorageProviderSettingsBase
@@ -15,6 +16,23 @@
 from snakemake_interface_storage_plugins.io import IOCacheStorageInterface
 
 
+def is_valid_azure_storage_blob_endpoint(endpoint_url: str) -> bool:
+    """
+    Validates if the blob account endpoint is a valid Azure Storage Account URL.
+
+    Args:
+    blob_account_url (str): The name of the environment variable.
+
+    Returns:
+    bool: True if the environment variable is a valid Azure Storage Account URL.
+    """
+    url_pattern = re.compile(
+        r"^https:\/\/[a-z0-9]+(\.[a-z0-9]+)*\.blob\.core\.windows\.net\/?(.+)?$"
+    )
+
+    return bool(url_pattern.match(endpoint_url))
+
+
 # Optional:
 # Define settings for your storage plugin (e.g. host url, credentials).
 # They will occur in the Snakemake CLI as --storage-<storage-plugin-name>-<param-name>
@@ -53,7 +71,10 @@ class StorageProviderSettings(StorageProviderSettingsBase):
     access_key: Optional[str] = field(
         default=None,
         metadata={
-            "help": "Azure Blob Storage Account Access Key Credential",
+            "help": (
+                "Azure Blob Storage Account Access Key Credential.",
+                "If set, takes precedence over sas_token credential.",
+            ),
             "env_var": False,
         },
     )
@@ -65,6 +86,18 @@ class StorageProviderSettings(StorageProviderSettingsBase):
         },
     )
 
+    def __post_init__(self):
+        if not is_valid_azure_storage_blob_endpoint(self.endpoint_url):
+            raise ValueError(
+                f"Invalid Azure Storage Blob Endpoint URL: {self.endpoint_url}"
+            )
+
+        self.credential = None
+        if self.access_key:
+            self.credential = self.access_key
+        elif self.sas_token:
+            self.credential = self.sas_token
+
 
 # Required:
 # Implementation of your storage provider