prefect_azure

AzureBlobStorageContainer

Bases: ObjectStorageBlock, WritableFileSystem, WritableDeploymentStorage

Represents a container in Azure Blob Storage.

This class provides methods for downloading and uploading files and folders to and from the Azure Blob Storage container.

Attributes:

Name	Type	Description
container_name	str	The name of the Azure Blob Storage container.
credentials	AzureBlobStorageCredentials	The credentials to use for authentication with Azure.
base_folder	Optional[str]	A base path to a folder within the container to use for reading and writing objects.

Source code in prefect_azure/blob_storage.py

class AzureBlobStorageContainer(
    ObjectStorageBlock, WritableFileSystem, WritableDeploymentStorage
):
    """
    Represents a container in Azure Blob Storage.

    This class provides methods for downloading and uploading files and folders
    to and from the Azure Blob Storage container.

    Attributes:
        container_name: The name of the Azure Blob Storage container.
        credentials: The credentials to use for authentication with Azure.
        base_folder: A base path to a folder within the container to use
            for reading and writing objects.
    """

    _block_type_name = "Azure Blob Storage Container"
    _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/54e3fa7e00197a4fbd1d82ed62494cb58d08c96a-250x250.png"  # noqa
    _documentation_url = "https://prefecthq.github.io/prefect-azure/blob_storage/#prefect_azure.blob_storabe.AzureBlobStorageContainer"  # noqa

    container_name: str = Field(
        default=..., description="The name of a Azure Blob Storage container."
    )
    credentials: AzureBlobStorageCredentials = Field(
        default_factory=AzureBlobStorageCredentials,
        description="The credentials to use for authentication with Azure.",
    )
    base_folder: Optional[str] = Field(
        default=None,
        description=(
            "A base path to a folder within the container to use "
            "for reading and writing objects."
        ),
    )

    def _get_path_relative_to_base_folder(self, path: Optional[str] = None) -> str:
        if path is None and self.base_folder is None:
            return ""
        if path is None:
            return self.base_folder
        if self.base_folder is None:
            return path
        return (Path(self.base_folder) / Path(path)).as_posix()

    @sync_compatible
    async def download_folder_to_path(
        self,
        from_folder: str,
        to_folder: Union[str, Path],
        **download_kwargs: Dict[str, Any],
    ) -> Coroutine[Any, Any, Path]:
        """Download a folder from the container to a local path.

        Args:
            from_folder: The folder path in the container to download.
            to_folder: The local path to download the folder to.
            **download_kwargs: Additional keyword arguments passed into
                `BlobClient.download_blob`.

        Returns:
            The local path where the folder was downloaded.

        Example:
            Download the contents of container folder `folder` from the container
                to the local folder `local_folder`:

            ```python
            from prefect_azure import AzureBlobStorageCredentials
            from prefect_azure.blob_storage import AzureBlobStorageContainer

            credentials = AzureBlobStorageCredentials(
                connection_string="connection_string",
            )
            block = AzureBlobStorageContainer(
                container_name="container",
                credentials=credentials,
            )
            block.download_folder_to_path(
                from_folder="folder",
                to_folder="local_folder"
            )
            ```
        """
        self.logger.info(
            "Downloading folder from container %s to path %s",
            self.container_name,
            to_folder,
        )
        full_container_path = self._get_path_relative_to_base_folder(from_folder)
        async with self.credentials.get_container_client(
            self.container_name
        ) as container_client:
            try:
                async for blob in container_client.list_blobs(
                    name_starts_with=full_container_path
                ):
                    blob_path = blob.name
                    local_path = Path(to_folder) / Path(blob_path).relative_to(
                        full_container_path
                    )
                    local_path.parent.mkdir(parents=True, exist_ok=True)
                    async with container_client.get_blob_client(
                        blob_path
                    ) as blob_client:
                        blob_obj = await blob_client.download_blob(**download_kwargs)

                    with local_path.open(mode="wb") as to_file:
                        await blob_obj.readinto(to_file)
            except ResourceNotFoundError as exc:
                raise RuntimeError(
                    "An error occurred when attempting to download from container"
                    f" {self.container_name}: {exc.reason}"
                ) from exc

        return Path(to_folder)

    @sync_compatible
    async def download_object_to_file_object(
        self,
        from_path: str,
        to_file_object: BinaryIO,
        **download_kwargs: Dict[str, Any],
    ) -> Coroutine[Any, Any, BinaryIO]:
        """
        Downloads an object from the container to a file object.

        Args:
            from_path : The path of the object to download within the container.
            to_file_object: The file object to download the object to.
            **download_kwargs: Additional keyword arguments for the download
                operation.

        Returns:
            The file object that the object was downloaded to.

        Example:
            Download the object `object` from the container to a file object:

            ```python
            from prefect_azure import AzureBlobStorageCredentials
            from prefect_azure.blob_storage import AzureBlobStorageContainer

            credentials = AzureBlobStorageCredentials(
                connection_string="connection_string",
            )
            block = AzureBlobStorageContainer(
                container_name="container",
                credentials=credentials,
            )
            with open("file.txt", "wb") as f:
                block.download_object_to_file_object(
                    from_path="object",
                    to_file_object=f
                )
            ```
        """
        self.logger.info(
            "Downloading object from container %s to file object", self.container_name
        )
        full_container_path = self._get_path_relative_to_base_folder(from_path)
        async with self.credentials.get_blob_client(
            self.container_name, full_container_path
        ) as blob_client:
            try:
                blob_obj = await blob_client.download_blob(**download_kwargs)
                await blob_obj.download_to_stream(to_file_object)
            except ResourceNotFoundError as exc:
                raise RuntimeError(
                    "An error occurred when attempting to download from container"
                    f" {self.container_name}: {exc.reason}"
                ) from exc

        return to_file_object

    @sync_compatible
    async def download_object_to_path(
        self,
        from_path: str,
        to_path: Union[str, Path],
        **download_kwargs: Dict[str, Any],
    ) -> Coroutine[Any, Any, Path]:
        """
        Downloads an object from a container to a specified path.

        Args:
            from_path: The path of the object in the container.
            to_path: The path where the object will be downloaded to.
            **download_kwargs (Dict[str, Any]): Additional keyword arguments
                for the download operation.

        Returns:
            The path where the object was downloaded to.

        Example:
            Download the object `object` from the container to the local path
                `file.txt`:

            ```python
            from prefect_azure import AzureBlobStorageCredentials
            from prefect_azure.blob_storage import AzureBlobStorageContainer

            credentials = AzureBlobStorageCredentials(
                connection_string="connection_string",
            )
            block = AzureBlobStorageContainer(
                container_name="container",
                credentials=credentials,
            )
            block.download_object_to_path(
                from_path="object",
                to_path="file.txt"
            )
            ```
        """
        self.logger.info(
            "Downloading object from container %s to path %s",
            self.container_name,
            to_path,
        )
        full_container_path = self._get_path_relative_to_base_folder(from_path)
        async with self.credentials.get_blob_client(
            self.container_name, full_container_path
        ) as blob_client:
            try:
                blob_obj = await blob_client.download_blob(**download_kwargs)

                path = Path(to_path)

                path.parent.mkdir(parents=True, exist_ok=True)

                with path.open(mode="wb") as to_file:
                    await blob_obj.readinto(to_file)
            except ResourceNotFoundError as exc:
                raise RuntimeError(
                    "An error occurred when attempting to download from container"
                    f" {self.container_name}: {exc.reason}"
                ) from exc
        return Path(to_path)

    @sync_compatible
    async def upload_from_file_object(
        self, from_file_object: BinaryIO, to_path: str, **upload_kwargs: Dict[str, Any]
    ) -> Coroutine[Any, Any, str]:
        """
        Uploads an object from a file object to the specified path in the blob
            storage container.

        Args:
            from_file_object: The file object to upload.
            to_path: The path in the blob storage container to upload the
                object to.
            **upload_kwargs: Additional keyword arguments to pass to the
                upload_blob method.

        Returns:
            The path where the object was uploaded to.

        Example:
            Upload a file object to the container at the path `object`:

            ```python
            from prefect_azure import AzureBlobStorageCredentials
            from prefect_azure.blob_storage import AzureBlobStorageContainer

            credentials = AzureBlobStorageCredentials(
                connection_string="connection_string",
            )
            block = AzureBlobStorageContainer(
                container_name="container",
                credentials=credentials,
            )
            with open("file.txt", "rb") as f:
                block.upload_from_file_object(
                    from_file_object=f,
                    to_path="object"
                )
            ```
        """
        self.logger.info(
            "Uploading object to container %s with key %s", self.container_name, to_path
        )
        full_container_path = self._get_path_relative_to_base_folder(to_path)
        async with self.credentials.get_blob_client(
            self.container_name, full_container_path
        ) as blob_client:
            try:
                await blob_client.upload_blob(from_file_object, **upload_kwargs)
            except ResourceNotFoundError as exc:
                raise RuntimeError(
                    "An error occurred when attempting to upload from container"
                    f" {self.container_name}: {exc.reason}"
                ) from exc

        return to_path

    @sync_compatible
    async def upload_from_path(
        self, from_path: Union[str, Path], to_path: str, **upload_kwargs: Dict[str, Any]
    ) -> Coroutine[Any, Any, str]:
        """
        Uploads an object from a local path to the specified destination path in the
            blob storage container.

        Args:
            from_path: The local path of the object to upload.
            to_path: The destination path in the blob storage container.
            **upload_kwargs: Additional keyword arguments to pass to the
                `upload_blob` method.

        Returns:
            The destination path in the blob storage container.

        Example:
            Upload a file from the local path `file.txt` to the container
                at the path `object`:

            ```python
            from prefect_azure import AzureBlobStorageCredentials
            from prefect_azure.blob_storage import AzureBlobStorageContainer

            credentials = AzureBlobStorageCredentials(
                connection_string="connection_string",
            )
            block = AzureBlobStorageContainer(
                container_name="container",
                credentials=credentials,
            )
            block.upload_from_path(
                from_path="file.txt",
                to_path="object"
            )
            ```
        """
        self.logger.info(
            "Uploading object to container %s with key %s", self.container_name, to_path
        )
        full_container_path = self._get_path_relative_to_base_folder(to_path)
        async with self.credentials.get_blob_client(
            self.container_name, full_container_path
        ) as blob_client:
            try:
                with open(from_path, "rb") as f:
                    await blob_client.upload_blob(f, **upload_kwargs)
            except ResourceNotFoundError as exc:
                raise RuntimeError(
                    "An error occurred when attempting to upload to container"
                    f" {self.container_name}: {exc.reason}"
                ) from exc

        return to_path

    @sync_compatible
    async def upload_from_folder(
        self,
        from_folder: Union[str, Path],
        to_folder: str,
        **upload_kwargs: Dict[str, Any],
    ) -> Coroutine[Any, Any, str]:
        """
        Uploads files from a local folder to a specified folder in the Azure
            Blob Storage container.

        Args:
            from_folder: The path to the local folder containing the files to upload.
            to_folder: The destination folder in the Azure Blob Storage container.
            **upload_kwargs: Additional keyword arguments to pass to the
                `upload_blob` method.

        Returns:
            The full path of the destination folder in the container.

        Example:
            Upload the contents of the local folder `local_folder` to the container
                folder `folder`:

            ```python
            from prefect_azure import AzureBlobStorageCredentials
            from prefect_azure.blob_storage import AzureBlobStorageContainer

            credentials = AzureBlobStorageCredentials(
                connection_string="connection_string",
            )
            block = AzureBlobStorageContainer(
                container_name="container",
                credentials=credentials,
            )
            block.upload_from_folder(
                from_folder="local_folder",
                to_folder="folder"
            )
            ```
        """
        self.logger.info(
            "Uploading folder to container %s with key %s",
            self.container_name,
            to_folder,
        )
        full_container_path = self._get_path_relative_to_base_folder(to_folder)
        async with self.credentials.get_container_client(
            self.container_name
        ) as container_client:
            if not Path(from_folder).is_dir():
                raise ValueError(f"{from_folder} is not a directory")
            for path in Path(from_folder).rglob("*"):
                if path.is_file():
                    blob_path = Path(full_container_path) / path.relative_to(
                        from_folder
                    )
                    async with container_client.get_blob_client(
                        blob_path.as_posix()
                    ) as blob_client:
                        try:
                            await blob_client.upload_blob(
                                path.read_bytes(), **upload_kwargs
                            )
                        except ResourceNotFoundError as exc:
                            raise RuntimeError(
                                "An error occurred when attempting to upload to "
                                f"container {self.container_name}: {exc.reason}"
                            ) from exc
        return full_container_path

    @sync_compatible
    async def get_directory(
        self, from_path: Optional[str] = None, local_path: Optional[str] = None
    ) -> None:
        """
        Downloads the contents of a direry from the blob storage to a local path.

        Used to enable flow code storage for deployments.

        Args:
            from_path: The path of the directory in the blob storage.
            local_path: The local path where the directory will be downloaded.
        """
        await self.download_folder_to_path(from_path, local_path)

    @sync_compatible
    async def put_directory(
        self,
        local_path: Optional[str] = None,
        to_path: Optional[str] = None,
        ignore_file: Optional[str] = None,
    ) -> None:
        """
        Uploads a directory to the blob storage.

        Used to enable flow code storage for deployments.

        Args:
            local_path: The local path of the directory to upload. Defaults to
                current directory.
            to_path: The destination path in the blob storage. Defaults to
                root directory.
            ignore_file: The path to a file containing patterns to ignore
                during upload.
        """
        to_path = "" if to_path is None else to_path

        if local_path is None:
            local_path = "."

        included_files = None
        if ignore_file:
            with open(ignore_file, "r") as f:
                ignore_patterns = f.readlines()

            included_files = filter_files(local_path, ignore_patterns)

        for local_file_path in Path(local_path).expanduser().rglob("*"):
            if (
                included_files is not None
                and str(local_file_path.relative_to(local_path)) not in included_files
            ):
                continue
            elif not local_file_path.is_dir():
                remote_file_path = Path(to_path) / local_file_path.relative_to(
                    local_path
                )
                with open(local_file_path, "rb") as local_file:
                    local_file_content = local_file.read()

                await self.write_path(
                    remote_file_path.as_posix(), content=local_file_content
                )

    @sync_compatible
    async def read_path(self, path: str) -> bytes:
        """
        Reads the contents of a file at the specified path and returns it as bytes.

        Used to enable results storage.

        Args:
            path: The path of the file to read.

        Returns:
            The contents of the file as bytes.
        """
        file_obj = BytesIO()
        await self.download_object_to_file_object(path, file_obj)
        return file_obj.getvalue()

    @sync_compatible
    async def write_path(self, path: str, content: bytes) -> None:
        """
        Writes the content to the specified path in the blob storage.

        Used to enable results storage.

        Args:
            path: The path where the content will be written.
            content: The content to be written.
        """
        await self.upload_from_file_object(BytesIO(content), path)

download_folder_to_path(from_folder, to_folder, **download_kwargs) async

Download a folder from the container to a local path.

Parameters:

Name	Type	Description	Default
from_folder	str	The folder path in the container to download.	required
to_folder	Union[str, Path]	The local path to download the folder to.	required
**download_kwargs	Dict[str, Any]	Additional keyword arguments passed into BlobClient.download_blob.	{}

Returns:

Type	Description
Coroutine[Any, Any, Path]	The local path where the folder was downloaded.

Example

Download the contents of container folder folder from the container to the local folder local_folder:

from prefect_azure import AzureBlobStorageCredentials
from prefect_azure.blob_storage import AzureBlobStorageContainer

credentials = AzureBlobStorageCredentials(
    connection_string="connection_string",
)
block = AzureBlobStorageContainer(
    container_name="container",
    credentials=credentials,
)
block.download_folder_to_path(
    from_folder="folder",
    to_folder="local_folder"
)

Source code in prefect_azure/blob_storage.py

@sync_compatible
async def download_folder_to_path(
    self,
    from_folder: str,
    to_folder: Union[str, Path],
    **download_kwargs: Dict[str, Any],
) -> Coroutine[Any, Any, Path]:
    """Download a folder from the container to a local path.

    Args:
        from_folder: The folder path in the container to download.
        to_folder: The local path to download the folder to.
        **download_kwargs: Additional keyword arguments passed into
            `BlobClient.download_blob`.

    Returns:
        The local path where the folder was downloaded.

    Example:
        Download the contents of container folder `folder` from the container
            to the local folder `local_folder`:

        ```python
        from prefect_azure import AzureBlobStorageCredentials
        from prefect_azure.blob_storage import AzureBlobStorageContainer

        credentials = AzureBlobStorageCredentials(
            connection_string="connection_string",
        )
        block = AzureBlobStorageContainer(
            container_name="container",
            credentials=credentials,
        )
        block.download_folder_to_path(
            from_folder="folder",
            to_folder="local_folder"
        )
        ```
    """
    self.logger.info(
        "Downloading folder from container %s to path %s",
        self.container_name,
        to_folder,
    )
    full_container_path = self._get_path_relative_to_base_folder(from_folder)
    async with self.credentials.get_container_client(
        self.container_name
    ) as container_client:
        try:
            async for blob in container_client.list_blobs(
                name_starts_with=full_container_path
            ):
                blob_path = blob.name
                local_path = Path(to_folder) / Path(blob_path).relative_to(
                    full_container_path
                )
                local_path.parent.mkdir(parents=True, exist_ok=True)
                async with container_client.get_blob_client(
                    blob_path
                ) as blob_client:
                    blob_obj = await blob_client.download_blob(**download_kwargs)

                with local_path.open(mode="wb") as to_file:
                    await blob_obj.readinto(to_file)
        except ResourceNotFoundError as exc:
            raise RuntimeError(
                "An error occurred when attempting to download from container"
                f" {self.container_name}: {exc.reason}"
            ) from exc

    return Path(to_folder)

download_object_to_file_object(from_path, to_file_object, **download_kwargs) async

Downloads an object from the container to a file object.

Parameters:

Name	Type	Description	Default
from_path		The path of the object to download within the container.	required
to_file_object	BinaryIO	The file object to download the object to.	required
**download_kwargs	Dict[str, Any]	Additional keyword arguments for the download operation.	{}

Returns:

Type	Description
Coroutine[Any, Any, BinaryIO]	The file object that the object was downloaded to.

Example

Download the object object from the container to a file object:

from prefect_azure import AzureBlobStorageCredentials
from prefect_azure.blob_storage import AzureBlobStorageContainer

credentials = AzureBlobStorageCredentials(
    connection_string="connection_string",
)
block = AzureBlobStorageContainer(
    container_name="container",
    credentials=credentials,
)
with open("file.txt", "wb") as f:
    block.download_object_to_file_object(
        from_path="object",
        to_file_object=f
    )

Source code in prefect_azure/blob_storage.py

@sync_compatible
async def download_object_to_file_object(
    self,
    from_path: str,
    to_file_object: BinaryIO,
    **download_kwargs: Dict[str, Any],
) -> Coroutine[Any, Any, BinaryIO]:
    """
    Downloads an object from the container to a file object.

    Args:
        from_path : The path of the object to download within the container.
        to_file_object: The file object to download the object to.
        **download_kwargs: Additional keyword arguments for the download
            operation.

    Returns:
        The file object that the object was downloaded to.

    Example:
        Download the object `object` from the container to a file object:

        ```python
        from prefect_azure import AzureBlobStorageCredentials
        from prefect_azure.blob_storage import AzureBlobStorageContainer

        credentials = AzureBlobStorageCredentials(
            connection_string="connection_string",
        )
        block = AzureBlobStorageContainer(
            container_name="container",
            credentials=credentials,
        )
        with open("file.txt", "wb") as f:
            block.download_object_to_file_object(
                from_path="object",
                to_file_object=f
            )
        ```
    """
    self.logger.info(
        "Downloading object from container %s to file object", self.container_name
    )
    full_container_path = self._get_path_relative_to_base_folder(from_path)
    async with self.credentials.get_blob_client(
        self.container_name, full_container_path
    ) as blob_client:
        try:
            blob_obj = await blob_client.download_blob(**download_kwargs)
            await blob_obj.download_to_stream(to_file_object)
        except ResourceNotFoundError as exc:
            raise RuntimeError(
                "An error occurred when attempting to download from container"
                f" {self.container_name}: {exc.reason}"
            ) from exc

    return to_file_object

download_object_to_path(from_path, to_path, **download_kwargs) async

Downloads an object from a container to a specified path.

Parameters:

Name	Type	Description	Default
from_path	str	The path of the object in the container.	required
to_path	Union[str, Path]	The path where the object will be downloaded to.	required
**download_kwargs	Dict[str, Any]	Additional keyword arguments for the download operation.	{}

Returns:

Type	Description
Coroutine[Any, Any, Path]	The path where the object was downloaded to.

Example

Download the object object from the container to the local path file.txt:

from prefect_azure import AzureBlobStorageCredentials
from prefect_azure.blob_storage import AzureBlobStorageContainer

credentials = AzureBlobStorageCredentials(
    connection_string="connection_string",
)
block = AzureBlobStorageContainer(
    container_name="container",
    credentials=credentials,
)
block.download_object_to_path(
    from_path="object",
    to_path="file.txt"
)

Source code in prefect_azure/blob_storage.py

@sync_compatible
async def download_object_to_path(
    self,
    from_path: str,
    to_path: Union[str, Path],
    **download_kwargs: Dict[str, Any],
) -> Coroutine[Any, Any, Path]:
    """
    Downloads an object from a container to a specified path.

    Args:
        from_path: The path of the object in the container.
        to_path: The path where the object will be downloaded to.
        **download_kwargs (Dict[str, Any]): Additional keyword arguments
            for the download operation.

    Returns:
        The path where the object was downloaded to.

    Example:
        Download the object `object` from the container to the local path
            `file.txt`:

        ```python
        from prefect_azure import AzureBlobStorageCredentials
        from prefect_azure.blob_storage import AzureBlobStorageContainer

        credentials = AzureBlobStorageCredentials(
            connection_string="connection_string",
        )
        block = AzureBlobStorageContainer(
            container_name="container",
            credentials=credentials,
        )
        block.download_object_to_path(
            from_path="object",
            to_path="file.txt"
        )
        ```
    """
    self.logger.info(
        "Downloading object from container %s to path %s",
        self.container_name,
        to_path,
    )
    full_container_path = self._get_path_relative_to_base_folder(from_path)
    async with self.credentials.get_blob_client(
        self.container_name, full_container_path
    ) as blob_client:
        try:
            blob_obj = await blob_client.download_blob(**download_kwargs)

            path = Path(to_path)

            path.parent.mkdir(parents=True, exist_ok=True)

            with path.open(mode="wb") as to_file:
                await blob_obj.readinto(to_file)
        except ResourceNotFoundError as exc:
            raise RuntimeError(
                "An error occurred when attempting to download from container"
                f" {self.container_name}: {exc.reason}"
            ) from exc
    return Path(to_path)

get_directory(from_path=None, local_path=None) async

Downloads the contents of a direry from the blob storage to a local path.

Used to enable flow code storage for deployments.

Parameters:

Name	Type	Description	Default
from_path	Optional[str]	The path of the directory in the blob storage.	None
local_path	Optional[str]	The local path where the directory will be downloaded.	None

Source code in prefect_azure/blob_storage.py

@sync_compatible
async def get_directory(
    self, from_path: Optional[str] = None, local_path: Optional[str] = None
) -> None:
    """
    Downloads the contents of a direry from the blob storage to a local path.

    Used to enable flow code storage for deployments.

    Args:
        from_path: The path of the directory in the blob storage.
        local_path: The local path where the directory will be downloaded.
    """
    await self.download_folder_to_path(from_path, local_path)

put_directory(local_path=None, to_path=None, ignore_file=None) async

Uploads a directory to the blob storage.

Used to enable flow code storage for deployments.

Parameters:

Name	Type	Description	Default
local_path	Optional[str]	The local path of the directory to upload. Defaults to current directory.	None
to_path	Optional[str]	The destination path in the blob storage. Defaults to root directory.	None
ignore_file	Optional[str]	The path to a file containing patterns to ignore during upload.	None

Source code in prefect_azure/blob_storage.py

@sync_compatible
async def put_directory(
    self,
    local_path: Optional[str] = None,
    to_path: Optional[str] = None,
    ignore_file: Optional[str] = None,
) -> None:
    """
    Uploads a directory to the blob storage.

    Used to enable flow code storage for deployments.

    Args:
        local_path: The local path of the directory to upload. Defaults to
            current directory.
        to_path: The destination path in the blob storage. Defaults to
            root directory.
        ignore_file: The path to a file containing patterns to ignore
            during upload.
    """
    to_path = "" if to_path is None else to_path

    if local_path is None:
        local_path = "."

    included_files = None
    if ignore_file:
        with open(ignore_file, "r") as f:
            ignore_patterns = f.readlines()

        included_files = filter_files(local_path, ignore_patterns)

    for local_file_path in Path(local_path).expanduser().rglob("*"):
        if (
            included_files is not None
            and str(local_file_path.relative_to(local_path)) not in included_files
        ):
            continue
        elif not local_file_path.is_dir():
            remote_file_path = Path(to_path) / local_file_path.relative_to(
                local_path
            )
            with open(local_file_path, "rb") as local_file:
                local_file_content = local_file.read()

            await self.write_path(
                remote_file_path.as_posix(), content=local_file_content
            )

read_path(path) async

Reads the contents of a file at the specified path and returns it as bytes.

Used to enable results storage.

Parameters:

Name	Type	Description	Default
path	str	The path of the file to read.	required

Returns:

Type	Description
bytes	The contents of the file as bytes.

Source code in prefect_azure/blob_storage.py

@sync_compatible
async def read_path(self, path: str) -> bytes:
    """
    Reads the contents of a file at the specified path and returns it as bytes.

    Used to enable results storage.

    Args:
        path: The path of the file to read.

    Returns:
        The contents of the file as bytes.
    """
    file_obj = BytesIO()
    await self.download_object_to_file_object(path, file_obj)
    return file_obj.getvalue()

upload_from_file_object(from_file_object, to_path, **upload_kwargs) async

Uploads an object from a file object to the specified path in the blob storage container.

Parameters:

Name	Type	Description	Default
from_file_object	BinaryIO	The file object to upload.	required
to_path	str	The path in the blob storage container to upload the object to.	required
**upload_kwargs	Dict[str, Any]	Additional keyword arguments to pass to the upload_blob method.	{}

Returns:

Type	Description
Coroutine[Any, Any, str]	The path where the object was uploaded to.

Example

Upload a file object to the container at the path object:

from prefect_azure import AzureBlobStorageCredentials
from prefect_azure.blob_storage import AzureBlobStorageContainer

credentials = AzureBlobStorageCredentials(
    connection_string="connection_string",
)
block = AzureBlobStorageContainer(
    container_name="container",
    credentials=credentials,
)
with open("file.txt", "rb") as f:
    block.upload_from_file_object(
        from_file_object=f,
        to_path="object"
    )

Source code in prefect_azure/blob_storage.py

@sync_compatible
async def upload_from_file_object(
    self, from_file_object: BinaryIO, to_path: str, **upload_kwargs: Dict[str, Any]
) -> Coroutine[Any, Any, str]:
    """
    Uploads an object from a file object to the specified path in the blob
        storage container.

    Args:
        from_file_object: The file object to upload.
        to_path: The path in the blob storage container to upload the
            object to.
        **upload_kwargs: Additional keyword arguments to pass to the
            upload_blob method.

    Returns:
        The path where the object was uploaded to.

    Example:
        Upload a file object to the container at the path `object`:

        ```python
        from prefect_azure import AzureBlobStorageCredentials
        from prefect_azure.blob_storage import AzureBlobStorageContainer

        credentials = AzureBlobStorageCredentials(
            connection_string="connection_string",
        )
        block = AzureBlobStorageContainer(
            container_name="container",
            credentials=credentials,
        )
        with open("file.txt", "rb") as f:
            block.upload_from_file_object(
                from_file_object=f,
                to_path="object"
            )
        ```
    """
    self.logger.info(
        "Uploading object to container %s with key %s", self.container_name, to_path
    )
    full_container_path = self._get_path_relative_to_base_folder(to_path)
    async with self.credentials.get_blob_client(
        self.container_name, full_container_path
    ) as blob_client:
        try:
            await blob_client.upload_blob(from_file_object, **upload_kwargs)
        except ResourceNotFoundError as exc:
            raise RuntimeError(
                "An error occurred when attempting to upload from container"
                f" {self.container_name}: {exc.reason}"
            ) from exc

    return to_path

upload_from_folder(from_folder, to_folder, **upload_kwargs) async

Uploads files from a local folder to a specified folder in the Azure Blob Storage container.

Parameters:

Name	Type	Description	Default
from_folder	Union[str, Path]	The path to the local folder containing the files to upload.	required
to_folder	str	The destination folder in the Azure Blob Storage container.	required
**upload_kwargs	Dict[str, Any]	Additional keyword arguments to pass to the upload_blob method.	{}

Returns:

Type	Description
Coroutine[Any, Any, str]	The full path of the destination folder in the container.

Example

Upload the contents of the local folder local_folder to the container folder folder:

from prefect_azure import AzureBlobStorageCredentials
from prefect_azure.blob_storage import AzureBlobStorageContainer

credentials = AzureBlobStorageCredentials(
    connection_string="connection_string",
)
block = AzureBlobStorageContainer(
    container_name="container",
    credentials=credentials,
)
block.upload_from_folder(
    from_folder="local_folder",
    to_folder="folder"
)

Source code in prefect_azure/blob_storage.py

@sync_compatible
async def upload_from_folder(
    self,
    from_folder: Union[str, Path],
    to_folder: str,
    **upload_kwargs: Dict[str, Any],
) -> Coroutine[Any, Any, str]:
    """
    Uploads files from a local folder to a specified folder in the Azure
        Blob Storage container.

    Args:
        from_folder: The path to the local folder containing the files to upload.
        to_folder: The destination folder in the Azure Blob Storage container.
        **upload_kwargs: Additional keyword arguments to pass to the
            `upload_blob` method.

    Returns:
        The full path of the destination folder in the container.

    Example:
        Upload the contents of the local folder `local_folder` to the container
            folder `folder`:

        ```python
        from prefect_azure import AzureBlobStorageCredentials
        from prefect_azure.blob_storage import AzureBlobStorageContainer

        credentials = AzureBlobStorageCredentials(
            connection_string="connection_string",
        )
        block = AzureBlobStorageContainer(
            container_name="container",
            credentials=credentials,
        )
        block.upload_from_folder(
            from_folder="local_folder",
            to_folder="folder"
        )
        ```
    """
    self.logger.info(
        "Uploading folder to container %s with key %s",
        self.container_name,
        to_folder,
    )
    full_container_path = self._get_path_relative_to_base_folder(to_folder)
    async with self.credentials.get_container_client(
        self.container_name
    ) as container_client:
        if not Path(from_folder).is_dir():
            raise ValueError(f"{from_folder} is not a directory")
        for path in Path(from_folder).rglob("*"):
            if path.is_file():
                blob_path = Path(full_container_path) / path.relative_to(
                    from_folder
                )
                async with container_client.get_blob_client(
                    blob_path.as_posix()
                ) as blob_client:
                    try:
                        await blob_client.upload_blob(
                            path.read_bytes(), **upload_kwargs
                        )
                    except ResourceNotFoundError as exc:
                        raise RuntimeError(
                            "An error occurred when attempting to upload to "
                            f"container {self.container_name}: {exc.reason}"
                        ) from exc
    return full_container_path

upload_from_path(from_path, to_path, **upload_kwargs) async

Uploads an object from a local path to the specified destination path in the blob storage container.

Parameters:

Name	Type	Description	Default
from_path	Union[str, Path]	The local path of the object to upload.	required
to_path	str	The destination path in the blob storage container.	required
**upload_kwargs	Dict[str, Any]	Additional keyword arguments to pass to the upload_blob method.	{}

Returns:

Type	Description
Coroutine[Any, Any, str]	The destination path in the blob storage container.

Example

Upload a file from the local path file.txt to the container at the path object:

from prefect_azure import AzureBlobStorageCredentials
from prefect_azure.blob_storage import AzureBlobStorageContainer

credentials = AzureBlobStorageCredentials(
    connection_string="connection_string",
)
block = AzureBlobStorageContainer(
    container_name="container",
    credentials=credentials,
)
block.upload_from_path(
    from_path="file.txt",
    to_path="object"
)

Source code in prefect_azure/blob_storage.py

@sync_compatible
async def upload_from_path(
    self, from_path: Union[str, Path], to_path: str, **upload_kwargs: Dict[str, Any]
) -> Coroutine[Any, Any, str]:
    """
    Uploads an object from a local path to the specified destination path in the
        blob storage container.

    Args:
        from_path: The local path of the object to upload.
        to_path: The destination path in the blob storage container.
        **upload_kwargs: Additional keyword arguments to pass to the
            `upload_blob` method.

    Returns:
        The destination path in the blob storage container.

    Example:
        Upload a file from the local path `file.txt` to the container
            at the path `object`:

        ```python
        from prefect_azure import AzureBlobStorageCredentials
        from prefect_azure.blob_storage import AzureBlobStorageContainer

        credentials = AzureBlobStorageCredentials(
            connection_string="connection_string",
        )
        block = AzureBlobStorageContainer(
            container_name="container",
            credentials=credentials,
        )
        block.upload_from_path(
            from_path="file.txt",
            to_path="object"
        )
        ```
    """
    self.logger.info(
        "Uploading object to container %s with key %s", self.container_name, to_path
    )
    full_container_path = self._get_path_relative_to_base_folder(to_path)
    async with self.credentials.get_blob_client(
        self.container_name, full_container_path
    ) as blob_client:
        try:
            with open(from_path, "rb") as f:
                await blob_client.upload_blob(f, **upload_kwargs)
        except ResourceNotFoundError as exc:
            raise RuntimeError(
                "An error occurred when attempting to upload to container"
                f" {self.container_name}: {exc.reason}"
            ) from exc

    return to_path

write_path(path, content) async

Writes the content to the specified path in the blob storage.

Used to enable results storage.

Parameters:

Name	Type	Description	Default
path	str	The path where the content will be written.	required
content	bytes	The content to be written.	required

Source code in prefect_azure/blob_storage.py

@sync_compatible
async def write_path(self, path: str, content: bytes) -> None:
    """
    Writes the content to the specified path in the blob storage.

    Used to enable results storage.

    Args:
        path: The path where the content will be written.
        content: The content to be written.
    """
    await self.upload_from_file_object(BytesIO(content), path)

AzureBlobStorageCredentials

Bases: Block

Stores credentials for authenticating with Azure Blob Storage.

Parameters:

Name	Type	Description	Default
account_url		The URL for your Azure storage account. If provided, the account URL will be used to authenticate with the discovered default Azure credentials.	required
connection_string		The connection string to your Azure storage account. If provided, the connection string will take precedence over the account URL.	required

Example

Load stored Azure Blob Storage credentials and retrieve a blob service client:

from prefect_azure import AzureBlobStorageCredentials

azure_credentials_block = AzureBlobStorageCredentials.load("BLOCK_NAME")

blob_service_client = azure_credentials_block.get_blob_client()

Source code in prefect_azure/credentials.py

class AzureBlobStorageCredentials(Block):
    """
    Stores credentials for authenticating with Azure Blob Storage.

    Args:
        account_url: The URL for your Azure storage account. If provided, the account
            URL will be used to authenticate with the discovered default Azure
            credentials.
        connection_string: The connection string to your Azure storage account. If
            provided, the connection string will take precedence over the account URL.

    Example:
        Load stored Azure Blob Storage credentials and retrieve a blob service client:
        ```python
        from prefect_azure import AzureBlobStorageCredentials

        azure_credentials_block = AzureBlobStorageCredentials.load("BLOCK_NAME")

        blob_service_client = azure_credentials_block.get_blob_client()
        ```
    """

    _block_type_name = "Azure Blob Storage Credentials"
    _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/54e3fa7e00197a4fbd1d82ed62494cb58d08c96a-250x250.png"  # noqa
    _documentation_url = "https://prefecthq.github.io/prefect-azure/credentials/#prefect_azure.credentials.AzureBlobStorageCredentials"  # noqa

    connection_string: Optional[SecretStr] = Field(
        default=None,
        description=(
            "The connection string to your Azure storage account. If provided, the "
            "connection string will take precedence over the account URL."
        ),
    )
    account_url: Optional[str] = Field(
        default=None,
        title="Account URL",
        description=(
            "The URL for your Azure storage account. If provided, the account "
            "URL will be used to authenticate with the discovered default "
            "Azure credentials."
        ),
    )

    @model_validator(mode="before")
    @classmethod
    def check_connection_string_or_account_url(
        cls, values: Dict[str, Any]
    ) -> Dict[str, Any]:
        """
        Checks that either a connection string or account URL is provided, not both.
        """
        has_account_url = values.get("account_url") is not None
        has_conn_str = values.get("connection_string") is not None
        if not has_account_url and not has_conn_str:
            raise ValueError(
                "Must provide either a connection string or an account URL."
            )
        if has_account_url and has_conn_str:
            raise ValueError(
                "Must provide either a connection string or account URL, but not both."
            )
        return values

    @_raise_help_msg("blob_storage")
    def get_client(self) -> "BlobServiceClient":
        """
        Returns an authenticated base Blob Service client that can be used to create
        other clients for Azure services.

        Example:
            Create an authorized Blob Service session
            ```python
            import os
            import asyncio
            from prefect import flow
            from prefect_azure import AzureBlobStorageCredentials

            @flow
            async def example_get_client_flow():
                connection_string = os.getenv("AZURE_STORAGE_CONNECTION_STRING")
                azure_credentials = AzureBlobStorageCredentials(
                    connection_string=connection_string,
                )
                async with azure_credentials.get_client() as blob_service_client:
                    # run other code here
                    pass

            asyncio.run(example_get_client_flow())
            ```
        """
        if self.connection_string is None:
            return BlobServiceClient(
                account_url=self.account_url,
                credential=ADefaultAzureCredential(),
            )

        return BlobServiceClient.from_connection_string(
            self.connection_string.get_secret_value()
        )

    @_raise_help_msg("blob_storage")
    def get_blob_client(self, container, blob) -> "BlobClient":
        """
        Returns an authenticated Blob client that can be used to
        download and upload blobs.

        Args:
            container: Name of the Blob Storage container to retrieve from.
            blob: Name of the blob within this container to retrieve.

        Example:
            Create an authorized Blob session
            ```python
            import os
            import asyncio
            from prefect import flow
            from prefect_azure import AzureBlobStorageCredentials

            @flow
            async def example_get_blob_client_flow():
                connection_string = os.getenv("AZURE_STORAGE_CONNECTION_STRING")
                azure_credentials = AzureBlobStorageCredentials(
                    connection_string=connection_string,
                )
                async with azure_credentials.get_blob_client(
                    "container", "blob"
                ) as blob_client:
                    # run other code here
                    pass

            asyncio.run(example_get_blob_client_flow())
            ```
        """
        if self.connection_string is None:
            return BlobClient(
                account_url=self.account_url,
                container_name=container,
                credential=ADefaultAzureCredential(),
                blob_name=blob,
            )

        blob_client = BlobClient.from_connection_string(
            self.connection_string.get_secret_value(), container, blob
        )
        return blob_client

    @_raise_help_msg("blob_storage")
    def get_container_client(self, container) -> "ContainerClient":
        """
        Returns an authenticated Container client that can be used to create clients
        for Azure services.

        Args:
            container: Name of the Blob Storage container to retrieve from.

        Example:
            Create an authorized Container session
            ```python
            import os
            import asyncio
            from prefect import flow
            from prefect_azure import AzureBlobStorageCredentials

            @flow
            async def example_get_container_client_flow():
                connection_string = os.getenv("AZURE_STORAGE_CONNECTION_STRING")
                azure_credentials = AzureBlobStorageCredentials(
                    connection_string=connection_string,
                )
                async with azure_credentials.get_container_client(
                    "container"
                ) as container_client:
                    # run other code here
                    pass

            asyncio.run(example_get_container_client_flow())
            ```
        """
        if self.connection_string is None:
            return ContainerClient(
                account_url=self.account_url,
                container_name=container,
                credential=ADefaultAzureCredential(),
            )

        container_client = ContainerClient.from_connection_string(
            self.connection_string.get_secret_value(), container
        )
        return container_client

check_connection_string_or_account_url(values) classmethod

Checks that either a connection string or account URL is provided, not both.

Source code in prefect_azure/credentials.py

@model_validator(mode="before")
@classmethod
def check_connection_string_or_account_url(
    cls, values: Dict[str, Any]
) -> Dict[str, Any]:
    """
    Checks that either a connection string or account URL is provided, not both.
    """
    has_account_url = values.get("account_url") is not None
    has_conn_str = values.get("connection_string") is not None
    if not has_account_url and not has_conn_str:
        raise ValueError(
            "Must provide either a connection string or an account URL."
        )
    if has_account_url and has_conn_str:
        raise ValueError(
            "Must provide either a connection string or account URL, but not both."
        )
    return values

get_blob_client(container, blob)

Returns an authenticated Blob client that can be used to download and upload blobs.

Parameters:

Name	Type	Description	Default
container		Name of the Blob Storage container to retrieve from.	required
blob		Name of the blob within this container to retrieve.	required

Example

Create an authorized Blob session

import os
import asyncio
from prefect import flow
from prefect_azure import AzureBlobStorageCredentials

@flow
async def example_get_blob_client_flow():
    connection_string = os.getenv("AZURE_STORAGE_CONNECTION_STRING")
    azure_credentials = AzureBlobStorageCredentials(
        connection_string=connection_string,
    )
    async with azure_credentials.get_blob_client(
        "container", "blob"
    ) as blob_client:
        # run other code here
        pass

asyncio.run(example_get_blob_client_flow())

Source code in prefect_azure/credentials.py

@_raise_help_msg("blob_storage")
def get_blob_client(self, container, blob) -> "BlobClient":
    """
    Returns an authenticated Blob client that can be used to
    download and upload blobs.

    Args:
        container: Name of the Blob Storage container to retrieve from.
        blob: Name of the blob within this container to retrieve.

    Example:
        Create an authorized Blob session
        ```python
        import os
        import asyncio
        from prefect import flow
        from prefect_azure import AzureBlobStorageCredentials

        @flow
        async def example_get_blob_client_flow():
            connection_string = os.getenv("AZURE_STORAGE_CONNECTION_STRING")
            azure_credentials = AzureBlobStorageCredentials(
                connection_string=connection_string,
            )
            async with azure_credentials.get_blob_client(
                "container", "blob"
            ) as blob_client:
                # run other code here
                pass

        asyncio.run(example_get_blob_client_flow())
        ```
    """
    if self.connection_string is None:
        return BlobClient(
            account_url=self.account_url,
            container_name=container,
            credential=ADefaultAzureCredential(),
            blob_name=blob,
        )

    blob_client = BlobClient.from_connection_string(
        self.connection_string.get_secret_value(), container, blob
    )
    return blob_client

get_client()

Returns an authenticated base Blob Service client that can be used to create other clients for Azure services.

Example

Create an authorized Blob Service session

import os
import asyncio
from prefect import flow
from prefect_azure import AzureBlobStorageCredentials

@flow
async def example_get_client_flow():
    connection_string = os.getenv("AZURE_STORAGE_CONNECTION_STRING")
    azure_credentials = AzureBlobStorageCredentials(
        connection_string=connection_string,
    )
    async with azure_credentials.get_client() as blob_service_client:
        # run other code here
        pass

asyncio.run(example_get_client_flow())

Source code in prefect_azure/credentials.py

@_raise_help_msg("blob_storage")
def get_client(self) -> "BlobServiceClient":
    """
    Returns an authenticated base Blob Service client that can be used to create
    other clients for Azure services.

    Example:
        Create an authorized Blob Service session
        ```python
        import os
        import asyncio
        from prefect import flow
        from prefect_azure import AzureBlobStorageCredentials

        @flow
        async def example_get_client_flow():
            connection_string = os.getenv("AZURE_STORAGE_CONNECTION_STRING")
            azure_credentials = AzureBlobStorageCredentials(
                connection_string=connection_string,
            )
            async with azure_credentials.get_client() as blob_service_client:
                # run other code here
                pass

        asyncio.run(example_get_client_flow())
        ```
    """
    if self.connection_string is None:
        return BlobServiceClient(
            account_url=self.account_url,
            credential=ADefaultAzureCredential(),
        )

    return BlobServiceClient.from_connection_string(
        self.connection_string.get_secret_value()
    )

get_container_client(container)

Returns an authenticated Container client that can be used to create clients for Azure services.

Parameters:

Name	Type	Description	Default
container		Name of the Blob Storage container to retrieve from.	required

Example

Create an authorized Container session

import os
import asyncio
from prefect import flow
from prefect_azure import AzureBlobStorageCredentials

@flow
async def example_get_container_client_flow():
    connection_string = os.getenv("AZURE_STORAGE_CONNECTION_STRING")
    azure_credentials = AzureBlobStorageCredentials(
        connection_string=connection_string,
    )
    async with azure_credentials.get_container_client(
        "container"
    ) as container_client:
        # run other code here
        pass

asyncio.run(example_get_container_client_flow())

Source code in prefect_azure/credentials.py

@_raise_help_msg("blob_storage")
def get_container_client(self, container) -> "ContainerClient":
    """
    Returns an authenticated Container client that can be used to create clients
    for Azure services.

    Args:
        container: Name of the Blob Storage container to retrieve from.

    Example:
        Create an authorized Container session
        ```python
        import os
        import asyncio
        from prefect import flow
        from prefect_azure import AzureBlobStorageCredentials

        @flow
        async def example_get_container_client_flow():
            connection_string = os.getenv("AZURE_STORAGE_CONNECTION_STRING")
            azure_credentials = AzureBlobStorageCredentials(
                connection_string=connection_string,
            )
            async with azure_credentials.get_container_client(
                "container"
            ) as container_client:
                # run other code here
                pass

        asyncio.run(example_get_container_client_flow())
        ```
    """
    if self.connection_string is None:
        return ContainerClient(
            account_url=self.account_url,
            container_name=container,
            credential=ADefaultAzureCredential(),
        )

    container_client = ContainerClient.from_connection_string(
        self.connection_string.get_secret_value(), container
    )
    return container_client

AzureContainerInstanceCredentials

Bases: Block

Block used to manage Azure Container Instances authentication. Stores Azure Service Principal authentication data.

Source code in prefect_azure/credentials.py

class AzureContainerInstanceCredentials(Block):
    """
    Block used to manage Azure Container Instances authentication. Stores Azure Service
    Principal authentication data.
    """

    _block_type_name = "Azure Container Instance Credentials"
    _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/54e3fa7e00197a4fbd1d82ed62494cb58d08c96a-250x250.png"  # noqa
    _documentation_url = "https://prefecthq.github.io/prefect-azure/credentials/#prefect_azure.credentials.AzureContainerInstanceCredentials"  # noqa

    client_id: Optional[str] = Field(
        default=None,
        title="Client ID",
        description=(
            "The service principal client ID. "
            "If none of client_id, tenant_id, and client_secret are provided, "
            "will use DefaultAzureCredential; else will need to provide all three to "
            "use ClientSecretCredential."
        ),
    )
    tenant_id: Optional[str] = Field(
        default=None,
        title="Tenant ID",
        description=(
            "The service principal tenant ID."
            "If none of client_id, tenant_id, and client_secret are provided, "
            "will use DefaultAzureCredential; else will need to provide all three to "
            "use ClientSecretCredential."
        ),
    )
    client_secret: Optional[SecretStr] = Field(
        default=None,
        description=(
            "The service principal client secret."
            "If none of client_id, tenant_id, and client_secret are provided, "
            "will use DefaultAzureCredential; else will need to provide all three to "
            "use ClientSecretCredential."
        ),
    )
    credential_kwargs: Dict[str, Any] = Field(
        default_factory=dict,
        title="Additional Credential Keyword Arguments",
        description=(
            "Additional keyword arguments to pass to "
            "`ClientSecretCredential` or `DefaultAzureCredential`."
        ),
    )

    @model_validator(mode="before")
    @classmethod
    def validate_credential_kwargs(cls, values):
        """
        Validates that if any of `client_id`, `tenant_id`, or `client_secret` are
        provided, all must be provided.
        """
        auth_args = ("client_id", "tenant_id", "client_secret")
        has_any = any(values.get(key) is not None for key in auth_args)
        has_all = all(values.get(key) is not None for key in auth_args)
        if has_any and not has_all:
            raise ValueError(
                "If any of `client_id`, `tenant_id`, or `client_secret` are provided, "
                "all must be provided."
            )
        return values

    def get_container_client(self, subscription_id: str):
        """
        Creates an Azure Container Instances client initialized with data from
        this block's fields and a provided Azure subscription ID.

        Args:
            subscription_id: A valid Azure subscription ID.

        Returns:
            An initialized `ContainerInstanceManagementClient`
        """

        return ContainerInstanceManagementClient(
            credential=self._create_credential(),
            subscription_id=subscription_id,
        )

    def get_resource_client(self, subscription_id: str):
        """
        Creates an Azure resource management client initialized with data from
        this block's fields and a provided Azure subscription ID.

        Args:
            subscription_id: A valid Azure subscription ID.

        Returns:
            An initialized `ResourceManagementClient`
        """

        return ResourceManagementClient(
            credential=self._create_credential(),
            subscription_id=subscription_id,
        )

    def _create_credential(self):
        """
        Creates an Azure credential initialized with data from this block's fields.

        Returns:
            An initialized Azure `TokenCredential` ready to use with Azure SDK client
            classes.
        """
        auth_args = (self.client_id, self.tenant_id, self.client_secret)
        if auth_args == (None, None, None):
            return DefaultAzureCredential(**self.credential_kwargs)

        return ClientSecretCredential(
            tenant_id=self.tenant_id,
            client_id=self.client_id,
            client_secret=self.client_secret.get_secret_value(),
            **self.credential_kwargs,
        )

get_container_client(subscription_id)

Creates an Azure Container Instances client initialized with data from this block's fields and a provided Azure subscription ID.

Parameters:

Name	Type	Description	Default
subscription_id	str	A valid Azure subscription ID.	required

Returns:

Type	Description
	An initialized ContainerInstanceManagementClient

Source code in prefect_azure/credentials.py

def get_container_client(self, subscription_id: str):
    """
    Creates an Azure Container Instances client initialized with data from
    this block's fields and a provided Azure subscription ID.

    Args:
        subscription_id: A valid Azure subscription ID.

    Returns:
        An initialized `ContainerInstanceManagementClient`
    """

    return ContainerInstanceManagementClient(
        credential=self._create_credential(),
        subscription_id=subscription_id,
    )

get_resource_client(subscription_id)

Creates an Azure resource management client initialized with data from this block's fields and a provided Azure subscription ID.

Parameters:

Name	Type	Description	Default
subscription_id	str	A valid Azure subscription ID.	required

Returns:

Type	Description
	An initialized ResourceManagementClient

Source code in prefect_azure/credentials.py

def get_resource_client(self, subscription_id: str):
    """
    Creates an Azure resource management client initialized with data from
    this block's fields and a provided Azure subscription ID.

    Args:
        subscription_id: A valid Azure subscription ID.

    Returns:
        An initialized `ResourceManagementClient`
    """

    return ResourceManagementClient(
        credential=self._create_credential(),
        subscription_id=subscription_id,
    )

validate_credential_kwargs(values) classmethod

Validates that if any of client_id, tenant_id, or client_secret are provided, all must be provided.

Source code in prefect_azure/credentials.py

@model_validator(mode="before")
@classmethod
def validate_credential_kwargs(cls, values):
    """
    Validates that if any of `client_id`, `tenant_id`, or `client_secret` are
    provided, all must be provided.
    """
    auth_args = ("client_id", "tenant_id", "client_secret")
    has_any = any(values.get(key) is not None for key in auth_args)
    has_all = all(values.get(key) is not None for key in auth_args)
    if has_any and not has_all:
        raise ValueError(
            "If any of `client_id`, `tenant_id`, or `client_secret` are provided, "
            "all must be provided."
        )
    return values

AzureContainerWorker

Bases: BaseWorker

A Prefect worker that runs flows in an Azure Container Instance.

Source code in prefect_azure/workers/container_instance.py

class AzureContainerWorker(BaseWorker):
    """
    A Prefect worker that runs flows in an Azure Container Instance.
    """

    type: str = "azure-container-instance"
    job_configuration = AzureContainerJobConfiguration
    job_configuration_variables = AzureContainerVariables
    _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/54e3fa7e00197a4fbd1d82ed62494cb58d08c96a-250x250.png"  # noqa
    _display_name = "Azure Container Instances"
    _description = (
        "Execute flow runs within containers on Azure's Container Instances "
        "service. Requires an Azure account."
    )
    _documentation_url = (
        "https://prefecthq.github.io/prefect-azure/container_instance_worker/"
    )

    async def run(
        self,
        flow_run: FlowRun,
        configuration: AzureContainerJobConfiguration,
        task_status: Optional[anyio.abc.TaskStatus] = None,
    ):
        """
        Run a flow in an Azure Container Instance.
        Args:
            flow_run: The flow run to run.
            configuration: The configuration for the flow run.
            task_status: The task status object for the current task. Used
            to provide an identifier that can be used to cancel the task.

        Returns:
            The result of the flow run.
        """
        run_start_time = datetime.datetime.now(datetime.timezone.utc)
        prefect_client = get_client()

        # Get the flow, so we can use its name in the container group name
        # to make it easier to identify and debug.
        flow = await prefect_client.read_flow(flow_run.flow_id)

        # Slugify flow.name and ensure that the generated name won't be too long
        # for the max deployment name length (64) including "prefect-"
        slugified_flow_name = slugify(
            flow.name,
            max_length=55 - len(str(flow_run.id)),
            regex_pattern=r"[^a-zA-Z0-9-]+",
        )
        container_group_name = f"{slugified_flow_name}-{flow_run.id}"

        self._logger.info(
            f"{self._log_prefix}: Preparing to run command {configuration.command} "
            f"in container  {configuration.image})..."
        )

        aci_client = configuration.aci_credentials.get_container_client(
            configuration.subscription_id.get_secret_value()
        )
        resource_client = configuration.aci_credentials.get_resource_client(
            configuration.subscription_id.get_secret_value()
        )

        created_container_group: Union[ContainerGroup, None] = None
        try:
            self._logger.info(f"{self._log_prefix}: Creating container group...")

            created_container_group = await self._provision_container_group(
                aci_client,
                resource_client,
                configuration,
                container_group_name,
            )
            # Both the flow ID and container group name will be needed to
            # cancel the flow run if needed.
            identifier = f"{flow_run.id}:{container_group_name}"

            if self._provisioning_succeeded(created_container_group):
                self._logger.info(f"{self._log_prefix}: Running command...")
                if task_status is not None:
                    task_status.started(value=identifier)

                status_code = await run_sync_in_worker_thread(
                    self._watch_task_and_get_exit_code,
                    aci_client,
                    configuration,
                    created_container_group,
                    run_start_time,
                )

                self._logger.info(f"{self._log_prefix}: Completed command run.")

            else:
                raise RuntimeError(f"{self._log_prefix}: Container creation failed.")

        finally:
            if configuration.keep_container_group:
                self._logger.info(f"{self._log_prefix}: Stopping container group...")
                aci_client.container_groups.stop(
                    resource_group_name=configuration.resource_group_name,
                    container_group_name=container_group_name,
                )
            else:
                await self._wait_for_container_group_deletion(
                    aci_client, configuration, container_group_name
                )

        return AzureContainerWorkerResult(
            identifier=created_container_group.name, status_code=status_code
        )

    def _wait_for_task_container_start(
        self,
        client: ContainerInstanceManagementClient,
        configuration: AzureContainerJobConfiguration,
        container_group_name: str,
        creation_status_poller: LROPoller[DeploymentExtended],
    ) -> Optional[ContainerGroup]:
        """
        Wait for the result of group and container creation.

        Args:
            creation_status_poller: Poller returned by the Azure SDK.

        Raises:
            RuntimeError: Raised if the timeout limit is exceeded before the
            container starts.

        Returns:
            A `ContainerGroup` representing the current status of the group being
            watched, or None if creation failed.
        """
        t0 = time.time()
        timeout = configuration.task_start_timeout_seconds

        while not creation_status_poller.done():
            elapsed_time = time.time() - t0

            if timeout and elapsed_time > timeout:
                raise RuntimeError(
                    (
                        f"Timed out after {elapsed_time}s while watching waiting for "
                        "container start."
                    )
                )
            time.sleep(configuration.task_watch_poll_interval)

        deployment = creation_status_poller.result()

        provisioning_succeeded = (
            deployment.properties.provisioning_state
            == ContainerGroupProvisioningState.SUCCEEDED
        )

        if provisioning_succeeded:
            return self._get_container_group(
                client, configuration.resource_group_name, container_group_name
            )
        else:
            return None

    async def _provision_container_group(
        self,
        aci_client: ContainerInstanceManagementClient,
        resource_client: ResourceManagementClient,
        configuration: AzureContainerJobConfiguration,
        container_group_name: str,
    ):
        """
        Create a container group and wait for it to start.
        Args:
            aci_client: An authenticated ACI client.
            resource_client: An authenticated resource client.
            configuration: The job configuration.
            container_group_name: The name of the container group to create.

        Returns:
            A `ContainerGroup` representing the container group that was created.
        """
        properties = DeploymentProperties(
            mode=DeploymentMode.INCREMENTAL,
            template=configuration.arm_template,
            parameters={"container_group_name": {"value": container_group_name}},
        )
        deployment = Deployment(properties=properties)

        creation_status_poller = await run_sync_in_worker_thread(
            resource_client.deployments.begin_create_or_update,
            resource_group_name=configuration.resource_group_name,
            deployment_name=f"prefect-{container_group_name}",
            parameters=deployment,
        )

        created_container_group = await run_sync_in_worker_thread(
            self._wait_for_task_container_start,
            aci_client,
            configuration,
            container_group_name,
            creation_status_poller,
        )

        return created_container_group

    def _watch_task_and_get_exit_code(
        self,
        client: ContainerInstanceManagementClient,
        configuration: AzureContainerJobConfiguration,
        container_group: ContainerGroup,
        run_start_time: datetime.datetime,
    ) -> int:
        """
        Waits until the container finishes running and obtains its exit code.

        Args:
            client: An initialized Azure `ContainerInstanceManagementClient`
            container_group: The `ContainerGroup` in which the container resides.

        Returns:
            An `int` representing the container's exit code.
        """
        status_code = -1
        running_container = self._get_container(container_group)
        current_state = running_container.instance_view.current_state.state

        # get any logs the container has already generated
        last_log_time = run_start_time
        if configuration.stream_output:
            last_log_time = self._get_and_stream_output(
                client=client,
                configuration=configuration,
                container_group=container_group,
                last_log_time=last_log_time,
            )

        # set exit code if flow run already finished:
        if current_state == ContainerRunState.TERMINATED:
            status_code = running_container.instance_view.current_state.exit_code

        while current_state != ContainerRunState.TERMINATED:
            try:
                container_group = self._get_container_group(
                    client,
                    configuration.resource_group_name,
                    container_group.name,
                )
            except ResourceNotFoundError:
                self._logger.exception(
                    f"{self._log_prefix}: Container group was deleted before flow run "
                    "completed, likely due to flow cancellation."
                )

                # since the flow was cancelled, exit early instead of raising an
                # exception
                return status_code

            container = self._get_container(container_group)
            current_state = container.instance_view.current_state.state

            if current_state == ContainerRunState.TERMINATED:
                status_code = container.instance_view.current_state.exit_code
                # break instead of waiting for next loop iteration because
                # trying to read logs from a terminated container raises an exception
                break

            if configuration.stream_output:
                last_log_time = self._get_and_stream_output(
                    client=client,
                    configuration=configuration,
                    container_group=container_group,
                    last_log_time=last_log_time,
                )

            time.sleep(configuration.task_watch_poll_interval)

        return status_code

    async def _wait_for_container_group_deletion(
        self,
        aci_client: ContainerInstanceManagementClient,
        configuration: AzureContainerJobConfiguration,
        container_group_name: str,
    ):
        """
        Wait for the container group to be deleted.
        Args:
            aci_client: An authenticated ACI client.
            configuration: The job configuration.
            container_group_name: The name of the container group to delete.
        """
        self._logger.info(f"{self._log_prefix}: Deleting container...")

        deletion_status_poller = await run_sync_in_worker_thread(
            aci_client.container_groups.begin_delete,
            resource_group_name=configuration.resource_group_name,
            container_group_name=container_group_name,
        )

        t0 = time.time()
        timeout = CONTAINER_GROUP_DELETION_TIMEOUT_SECONDS

        while not deletion_status_poller.done():
            elapsed_time = time.time() - t0

            if timeout and elapsed_time > timeout:
                raise RuntimeError(
                    (
                        f"Timed out after {elapsed_time}s while waiting for deletion of"
                        f" container group {container_group_name}. To verify the group "
                        "has been deleted, check the Azure Portal or run "
                        f"az container show --name {container_group_name} --resource-group {configuration.resource_group_name}"  # noqa
                    )
                )
            await anyio.sleep(configuration.task_watch_poll_interval)

        self._logger.info(f"{self._log_prefix}: Container deleted.")

    def _get_container(self, container_group: ContainerGroup) -> Container:
        """
        Extracts the job container from a container group.
        """
        return container_group.containers[0]

    @staticmethod
    def _get_container_group(
        client: ContainerInstanceManagementClient,
        resource_group_name: str,
        container_group_name: str,
    ) -> ContainerGroup:
        """
        Gets the container group from Azure.
        """
        return client.container_groups.get(
            resource_group_name=resource_group_name,
            container_group_name=container_group_name,
        )

    def _get_and_stream_output(
        self,
        client: ContainerInstanceManagementClient,
        configuration: AzureContainerJobConfiguration,
        container_group: ContainerGroup,
        last_log_time: datetime.datetime,
    ) -> datetime.datetime:
        """
        Fetches logs output from the job container and writes all entries after
        a given time to stderr.

        Args:
            client: An initialized `ContainerInstanceManagementClient`
            container_group: The container group that holds the job container.
            last_log_time: The timestamp of the last output line already streamed.

        Returns:
            The time of the most recent output line written by this call.
        """
        logs = self._get_logs(
            client=client, configuration=configuration, container_group=container_group
        )
        return self._stream_output(logs, last_log_time)

    def _get_logs(
        self,
        client: ContainerInstanceManagementClient,
        configuration: AzureContainerJobConfiguration,
        container_group: ContainerGroup,
        max_lines: int = 100,
    ) -> str:
        """
        Gets the most container logs up to a given maximum.

        Args:
            client: An initialized `ContainerInstanceManagementClient`
            container_group: The container group that holds the job container.
            max_lines: The number of log lines to pull. Defaults to 100.

        Returns:
            A string containing the requested log entries, one per line.
        """
        container = self._get_container(container_group)

        logs: Union[Logs, None] = None
        try:
            logs = client.containers.list_logs(
                resource_group_name=configuration.resource_group_name,
                container_group_name=container_group.name,
                container_name=container.name,
                tail=max_lines,
                timestamps=True,
            )
        except HttpResponseError:
            # Trying to get logs when the container is under heavy CPU load sometimes
            # results in an error, but we won't want to raise an exception and stop
            # monitoring the flow. Instead, log the error and carry on so we can try to
            # get all missed logs on the next check.
            self._logger.warning(
                f"{self._log_prefix}: Unable to retrieve logs from container "
                f"{container.name}. Trying again in "
                f"{configuration.task_watch_poll_interval}s"
            )

        return logs.content if logs else ""

    def _stream_output(
        self, log_content: Union[str, None], last_log_time: datetime.datetime
    ) -> datetime.datetime:
        """
        Writes each entry from a string of log lines to stderr.

        Args:
            log_content: A string containing Azure container logs.
            last_log_time: The timestamp of the last output line already streamed.

        Returns:
            The time of the most recent output line written by this call.
        """
        if not log_content:
            # nothing to stream
            return last_log_time

        log_lines = log_content.split("\n")

        last_written_time = last_log_time

        for log_line in log_lines:
            # skip if the line is blank or whitespace
            if not log_line.strip():
                continue

            line_parts = log_line.split(" ")
            # timestamp should always be before first space in line
            line_timestamp = line_parts[0]
            line = " ".join(line_parts[1:])

            try:
                line_time = dateutil.parser.parse(line_timestamp)
                if line_time > last_written_time:
                    self._write_output_line(line)
                    last_written_time = line_time
            except dateutil.parser.ParserError as e:
                self._logger.debug(
                    (
                        f"{self._log_prefix}: Unable to parse timestamp from Azure "
                        "log line: %s"
                    ),
                    log_line,
                    exc_info=e,
                )

        return last_written_time

    @property
    def _log_prefix(self) -> str:
        """
        Internal property for generating a prefix for logs where `name` may be null
        """
        if self.name is not None:
            return f"AzureContainerInstanceJob {self.name!r}"
        else:
            return "AzureContainerInstanceJob"

    @staticmethod
    def _provisioning_succeeded(container_group: Union[ContainerGroup, None]) -> bool:
        """
        Determines whether ACI container group provisioning was successful.

        Args:
            container_group: a container group returned by the Azure SDK.

        Returns:
            True if provisioning was successful, False otherwise.
        """
        if not container_group:
            return False

        return (
            container_group.provisioning_state
            == ContainerGroupProvisioningState.SUCCEEDED
            and len(container_group.containers) == 1
        )

    @staticmethod
    def _write_output_line(line: str):
        """
        Writes a line of output to stderr.
        """
        print(line, file=sys.stderr)

run(flow_run, configuration, task_status=None) async

Run a flow in an Azure Container Instance. Args: flow_run: The flow run to run. configuration: The configuration for the flow run. task_status: The task status object for the current task. Used to provide an identifier that can be used to cancel the task.

Returns:

Type	Description
	The result of the flow run.

Source code in prefect_azure/workers/container_instance.py

async def run(
    self,
    flow_run: FlowRun,
    configuration: AzureContainerJobConfiguration,
    task_status: Optional[anyio.abc.TaskStatus] = None,
):
    """
    Run a flow in an Azure Container Instance.
    Args:
        flow_run: The flow run to run.
        configuration: The configuration for the flow run.
        task_status: The task status object for the current task. Used
        to provide an identifier that can be used to cancel the task.

    Returns:
        The result of the flow run.
    """
    run_start_time = datetime.datetime.now(datetime.timezone.utc)
    prefect_client = get_client()

    # Get the flow, so we can use its name in the container group name
    # to make it easier to identify and debug.
    flow = await prefect_client.read_flow(flow_run.flow_id)

    # Slugify flow.name and ensure that the generated name won't be too long
    # for the max deployment name length (64) including "prefect-"
    slugified_flow_name = slugify(
        flow.name,
        max_length=55 - len(str(flow_run.id)),
        regex_pattern=r"[^a-zA-Z0-9-]+",
    )
    container_group_name = f"{slugified_flow_name}-{flow_run.id}"

    self._logger.info(
        f"{self._log_prefix}: Preparing to run command {configuration.command} "
        f"in container  {configuration.image})..."
    )

    aci_client = configuration.aci_credentials.get_container_client(
        configuration.subscription_id.get_secret_value()
    )
    resource_client = configuration.aci_credentials.get_resource_client(
        configuration.subscription_id.get_secret_value()
    )

    created_container_group: Union[ContainerGroup, None] = None
    try:
        self._logger.info(f"{self._log_prefix}: Creating container group...")

        created_container_group = await self._provision_container_group(
            aci_client,
            resource_client,
            configuration,
            container_group_name,
        )
        # Both the flow ID and container group name will be needed to
        # cancel the flow run if needed.
        identifier = f"{flow_run.id}:{container_group_name}"

        if self._provisioning_succeeded(created_container_group):
            self._logger.info(f"{self._log_prefix}: Running command...")
            if task_status is not None:
                task_status.started(value=identifier)

            status_code = await run_sync_in_worker_thread(
                self._watch_task_and_get_exit_code,
                aci_client,
                configuration,
                created_container_group,
                run_start_time,
            )

            self._logger.info(f"{self._log_prefix}: Completed command run.")

        else:
            raise RuntimeError(f"{self._log_prefix}: Container creation failed.")

    finally:
        if configuration.keep_container_group:
            self._logger.info(f"{self._log_prefix}: Stopping container group...")
            aci_client.container_groups.stop(
                resource_group_name=configuration.resource_group_name,
                container_group_name=container_group_name,
            )
        else:
            await self._wait_for_container_group_deletion(
                aci_client, configuration, container_group_name
            )

    return AzureContainerWorkerResult(
        identifier=created_container_group.name, status_code=status_code
    )

AzureCosmosDbCredentials

Bases: Block

Block used to manage Cosmos DB authentication with Azure. Azure authentication is handled via the azure module through a connection string.

Parameters:

Name	Type	Description	Default
connection_string		Includes the authorization information required.	required

Example

Load stored Azure Cosmos DB credentials:

from prefect_azure import AzureCosmosDbCredentials
azure_credentials_block = AzureCosmosDbCredentials.load("BLOCK_NAME")

Source code in prefect_azure/credentials.py

class AzureCosmosDbCredentials(Block):
    """
    Block used to manage Cosmos DB authentication with Azure.
    Azure authentication is handled via the `azure` module through
    a connection string.

    Args:
        connection_string: Includes the authorization information required.

    Example:
        Load stored Azure Cosmos DB credentials:
        ```python
        from prefect_azure import AzureCosmosDbCredentials
        azure_credentials_block = AzureCosmosDbCredentials.load("BLOCK_NAME")
        ```
    """

    _block_type_name = "Azure Cosmos DB Credentials"
    _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/54e3fa7e00197a4fbd1d82ed62494cb58d08c96a-250x250.png"  # noqa
    _documentation_url = "https://prefecthq.github.io/prefect-azure/credentials/#prefect_azure.credentials.AzureCosmosDbCredentials"  # noqa

    connection_string: SecretStr = Field(
        default=..., description="Includes the authorization information required."
    )

    @_raise_help_msg("cosmos_db")
    def get_client(self) -> "CosmosClient":
        """
        Returns an authenticated Cosmos client that can be used to create
        other clients for Azure services.

        Example:
            Create an authorized Cosmos session
            ```python
            import os
            from prefect import flow
            from prefect_azure import AzureCosmosDbCredentials

            @flow
            def example_get_client_flow():
                connection_string = os.getenv("AZURE_COSMOS_CONNECTION_STRING")
                azure_credentials = AzureCosmosDbCredentials(
                    connection_string=connection_string,
                )
                cosmos_client = azure_credentials.get_client()
                return cosmos_client

            example_get_client_flow()
            ```
        """
        return CosmosClient.from_connection_string(
            self.connection_string.get_secret_value()
        )

    def get_database_client(self, database: str) -> "DatabaseProxy":
        """
        Returns an authenticated Database client.

        Args:
            database: Name of the database.

        Example:
            Create an authorized Cosmos session
            ```python
            import os
            from prefect import flow
            from prefect_azure import AzureCosmosDbCredentials

            @flow
            def example_get_client_flow():
                connection_string = os.getenv("AZURE_COSMOS_CONNECTION_STRING")
                azure_credentials = AzureCosmosDbCredentials(
                    connection_string=connection_string,
                )
                cosmos_client = azure_credentials.get_database_client()
                return cosmos_client

            example_get_database_client_flow()
            ```
        """
        cosmos_client = self.get_client()
        database_client = cosmos_client.get_database_client(database=database)
        return database_client

    def get_container_client(self, container: str, database: str) -> "ContainerProxy":
        """
        Returns an authenticated Container client used for querying.

        Args:
            container: Name of the Cosmos DB container to retrieve from.
            database: Name of the Cosmos DB database.

        Example:
            Create an authorized Container session
            ```python
            import os
            from prefect import flow
            from prefect_azure import AzureBlobStorageCredentials

            @flow
            def example_get_container_client_flow():
                connection_string = os.getenv("AZURE_COSMOS_CONNECTION_STRING")
                azure_credentials = AzureCosmosDbCredentials(
                    connection_string=connection_string,
                )
                container_client = azure_credentials.get_container_client(container)
                return container_client

            example_get_container_client_flow()
            ```
        """
        database_client = self.get_database_client(database)
        container_client = database_client.get_container_client(container=container)
        return container_client

get_client()

Returns an authenticated Cosmos client that can be used to create other clients for Azure services.

Example

Create an authorized Cosmos session

import os
from prefect import flow
from prefect_azure import AzureCosmosDbCredentials

@flow
def example_get_client_flow():
    connection_string = os.getenv("AZURE_COSMOS_CONNECTION_STRING")
    azure_credentials = AzureCosmosDbCredentials(
        connection_string=connection_string,
    )
    cosmos_client = azure_credentials.get_client()
    return cosmos_client

example_get_client_flow()

Source code in prefect_azure/credentials.py

@_raise_help_msg("cosmos_db")
def get_client(self) -> "CosmosClient":
    """
    Returns an authenticated Cosmos client that can be used to create
    other clients for Azure services.

    Example:
        Create an authorized Cosmos session
        ```python
        import os
        from prefect import flow
        from prefect_azure import AzureCosmosDbCredentials

        @flow
        def example_get_client_flow():
            connection_string = os.getenv("AZURE_COSMOS_CONNECTION_STRING")
            azure_credentials = AzureCosmosDbCredentials(
                connection_string=connection_string,
            )
            cosmos_client = azure_credentials.get_client()
            return cosmos_client

        example_get_client_flow()
        ```
    """
    return CosmosClient.from_connection_string(
        self.connection_string.get_secret_value()
    )

get_container_client(container, database)

Returns an authenticated Container client used for querying.

Parameters:

Name	Type	Description	Default
container	str	Name of the Cosmos DB container to retrieve from.	required
database	str	Name of the Cosmos DB database.	required

Example

Create an authorized Container session

import os
from prefect import flow
from prefect_azure import AzureBlobStorageCredentials

@flow
def example_get_container_client_flow():
    connection_string = os.getenv("AZURE_COSMOS_CONNECTION_STRING")
    azure_credentials = AzureCosmosDbCredentials(
        connection_string=connection_string,
    )
    container_client = azure_credentials.get_container_client(container)
    return container_client

example_get_container_client_flow()

Source code in prefect_azure/credentials.py

def get_container_client(self, container: str, database: str) -> "ContainerProxy":
    """
    Returns an authenticated Container client used for querying.

    Args:
        container: Name of the Cosmos DB container to retrieve from.
        database: Name of the Cosmos DB database.

    Example:
        Create an authorized Container session
        ```python
        import os
        from prefect import flow
        from prefect_azure import AzureBlobStorageCredentials

        @flow
        def example_get_container_client_flow():
            connection_string = os.getenv("AZURE_COSMOS_CONNECTION_STRING")
            azure_credentials = AzureCosmosDbCredentials(
                connection_string=connection_string,
            )
            container_client = azure_credentials.get_container_client(container)
            return container_client

        example_get_container_client_flow()
        ```
    """
    database_client = self.get_database_client(database)
    container_client = database_client.get_container_client(container=container)
    return container_client

get_database_client(database)

Returns an authenticated Database client.

Parameters:

Name	Type	Description	Default
database	str	Name of the database.	required

Example

Create an authorized Cosmos session

import os
from prefect import flow
from prefect_azure import AzureCosmosDbCredentials

@flow
def example_get_client_flow():
    connection_string = os.getenv("AZURE_COSMOS_CONNECTION_STRING")
    azure_credentials = AzureCosmosDbCredentials(
        connection_string=connection_string,
    )
    cosmos_client = azure_credentials.get_database_client()
    return cosmos_client

example_get_database_client_flow()

Source code in prefect_azure/credentials.py

def get_database_client(self, database: str) -> "DatabaseProxy":
    """
    Returns an authenticated Database client.

    Args:
        database: Name of the database.

    Example:
        Create an authorized Cosmos session
        ```python
        import os
        from prefect import flow
        from prefect_azure import AzureCosmosDbCredentials

        @flow
        def example_get_client_flow():
            connection_string = os.getenv("AZURE_COSMOS_CONNECTION_STRING")
            azure_credentials = AzureCosmosDbCredentials(
                connection_string=connection_string,
            )
            cosmos_client = azure_credentials.get_database_client()
            return cosmos_client

        example_get_database_client_flow()
        ```
    """
    cosmos_client = self.get_client()
    database_client = cosmos_client.get_database_client(database=database)
    return database_client

AzureMlCredentials

Bases: Block

Block used to manage authentication with AzureML. Azure authentication is handled via the azure module.

Parameters:

Name	Type	Description	Default
tenant_id		The active directory tenant that the service identity belongs to.	required
service_principal_id		The service principal ID.	required
service_principal_password		The service principal password/key.	required
subscription_id		The Azure subscription ID containing the workspace.	required
resource_group		The resource group containing the workspace.	required
workspace_name		The existing workspace name.	required

Example

Load stored AzureML credentials:

from prefect_azure import AzureMlCredentials
azure_ml_credentials_block = AzureMlCredentials.load("BLOCK_NAME")

Source code in prefect_azure/credentials.py

class AzureMlCredentials(Block):
    """
    Block used to manage authentication with AzureML. Azure authentication is
    handled via the `azure` module.

    Args:
        tenant_id: The active directory tenant that the service identity belongs to.
        service_principal_id: The service principal ID.
        service_principal_password: The service principal password/key.
        subscription_id: The Azure subscription ID containing the workspace.
        resource_group: The resource group containing the workspace.
        workspace_name: The existing workspace name.

    Example:
        Load stored AzureML credentials:
        ```python
        from prefect_azure import AzureMlCredentials
        azure_ml_credentials_block = AzureMlCredentials.load("BLOCK_NAME")
        ```
    """

    _block_type_name = "AzureML Credentials"
    _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/54e3fa7e00197a4fbd1d82ed62494cb58d08c96a-250x250.png"  # noqa
    _documentation_url = "https://prefecthq.github.io/prefect-azure/credentials/#prefect_azure.credentials.AzureMlCredentials"  # noqa

    tenant_id: str = Field(
        default=...,
        description="The active directory tenant that the service identity belongs to.",
    )
    service_principal_id: str = Field(
        default=..., description="The service principal ID."
    )
    service_principal_password: SecretStr = Field(
        default=..., description="The service principal password/key."
    )
    subscription_id: str = Field(
        default=...,
        description="The Azure subscription ID containing the workspace, in format: '00000000-0000-0000-0000-000000000000'.",  # noqa
    )
    resource_group: str = Field(
        default=..., description="The resource group containing the workspace."
    )
    workspace_name: str = Field(default=..., description="The existing workspace name.")

    @_raise_help_msg("ml_datastore")
    def get_workspace(self) -> "Workspace":
        """
        Returns an authenticated base Workspace that can be used in
        Azure's Datasets and Datastores.

        Example:
            Create an authorized workspace
            ```python
            import os
            from prefect import flow
            from prefect_azure import AzureMlCredentials
            @flow
            def example_get_workspace_flow():
                azure_credentials = AzureMlCredentials(
                    tenant_id="tenant_id",
                    service_principal_id="service_principal_id",
                    service_principal_password="service_principal_password",
                    subscription_id="subscription_id",
                    resource_group="resource_group",
                    workspace_name="workspace_name"
                )
                workspace_client = azure_credentials.get_workspace()
                return workspace_client
            example_get_workspace_flow()
            ```
        """
        service_principal_password = self.service_principal_password.get_secret_value()
        service_principal_authentication = ServicePrincipalAuthentication(
            tenant_id=self.tenant_id,
            service_principal_id=self.service_principal_id,
            service_principal_password=service_principal_password,
        )

        workspace = Workspace(
            subscription_id=self.subscription_id,
            resource_group=self.resource_group,
            workspace_name=self.workspace_name,
            auth=service_principal_authentication,
        )

        return workspace

get_workspace()

Returns an authenticated base Workspace that can be used in Azure's Datasets and Datastores.

Example

Create an authorized workspace

import os
from prefect import flow
from prefect_azure import AzureMlCredentials
@flow
def example_get_workspace_flow():
    azure_credentials = AzureMlCredentials(
        tenant_id="tenant_id",
        service_principal_id="service_principal_id",
        service_principal_password="service_principal_password",
        subscription_id="subscription_id",
        resource_group="resource_group",
        workspace_name="workspace_name"
    )
    workspace_client = azure_credentials.get_workspace()
    return workspace_client
example_get_workspace_flow()

Source code in prefect_azure/credentials.py

@_raise_help_msg("ml_datastore")
def get_workspace(self) -> "Workspace":
    """
    Returns an authenticated base Workspace that can be used in
    Azure's Datasets and Datastores.

    Example:
        Create an authorized workspace
        ```python
        import os
        from prefect import flow
        from prefect_azure import AzureMlCredentials
        @flow
        def example_get_workspace_flow():
            azure_credentials = AzureMlCredentials(
                tenant_id="tenant_id",
                service_principal_id="service_principal_id",
                service_principal_password="service_principal_password",
                subscription_id="subscription_id",
                resource_group="resource_group",
                workspace_name="workspace_name"
            )
            workspace_client = azure_credentials.get_workspace()
            return workspace_client
        example_get_workspace_flow()
        ```
    """
    service_principal_password = self.service_principal_password.get_secret_value()
    service_principal_authentication = ServicePrincipalAuthentication(
        tenant_id=self.tenant_id,
        service_principal_id=self.service_principal_id,
        service_principal_password=service_principal_password,
    )

    workspace = Workspace(
        subscription_id=self.subscription_id,
        resource_group=self.resource_group,
        workspace_name=self.workspace_name,
        auth=service_principal_authentication,
    )

    return workspace