image.png

pip install prefect-azure

and

pip install prefect-azure[blob-storage]

prefect block register -m prefect_azure

Then they will be available in the UI for you to configure

for this:

pip install prefect-azure[blob-storage]

image.png

We are currently at 2.10.20

The UI should be fine as those blocks are registered with the UI, but if you are trying to run and load a blob storage credential locally, it might fail due to the prefect_azure module being unable to load

Here’s an entire flow that uses that specific credential for reference:

from prefect import task, flow
from prefect import get_run_logger
import pandas as pd
import os
from io import BytesIO

from prefect_azure import AzureBlobStorageCredentials
from prefect_azure.blob_storage import blob_storage_download, blob_storage_upload


def azure_creds():

    try:
        azure_credentials_block = AzureBlobStorageCredentials.load("boydoblobbo")
        return azure_credentials_block
    except ValueError as e:
        get_run_logger().info(f"No azure_credentials_block found :{e}")
        try:
            connection_string = os.getenv("AZURE_STORAGE_CONNECTION_STRING")
            return AzureBlobStorageCredentials(connection_string=connection_string)
        except Exception as f:
            get_run_logger().info("No connection string found")
            connection_string = None
            raise


def load_from_azure():

    blob_storage_credentials = azure_creds()
    data = blob_storage_download(
        blob="file.csv",
        container="prefect-logs",
        blob_storage_credentials=blob_storage_credentials,
    )
    return data


@task
def read_file(data):
    return pd.read_csv(BytesIO(data))


def write_df(data):
    df = pd.DataFrame(data, columns=["output"])
    csv_data = df.to_csv()
    blob = blob_storage_upload(
        data=csv_data,
        container="prefect-logs",
        blob="csv_data",
        blob_storage_credentials=azure_creds(),
        overwrite=True,
    )
    return blob


@task
def transform_pd(df):
    results = [row["col1"] * row["col2"] for index, row in df.iterrows()]
    get_run_logger().info(f"{results=}")
    return results


@flow(log_prints=True)
def transform_flow():

    file = load_from_azure()
    df = read_file(file)
    transformed_output = transform_pd(df)
    write_df(transformed_output)


if __name__ == "__main__":
    transform_flow()

Much of this was re-factored as it was a bit difficult to use for many. The way to do this currently is either pythonically in code, or through prefect init / prefect deploy

you should be able to upgrade your prefect version

then run

prefect init

it will walk you through an interactive process

https://docs.prefect.io/2.10.18/tutorial/deployments/ It should be much more straight-forward -

prefect init

will allow you to choose from an interactive process. Since you’re using azure, it will setup your

prefect.yaml

for you You’ll likely still need to create an AzureBlobStorageCredential block which should just require a connection string Then the command would just be

prefect deploy

and it will walk you interactively through the process of your deployment - you can certainly pass them all via cli if you want and know them, but this makes it much simpler to follow

I know it probably seems like a lot to shift at the moment, but I assure you it’s much simpler now than it’s ever been and after the first one will be a lot more intuitive

with that done, you can just run

prefect deploy

provide your flow, and it will go to the storage you selected and pull from the storage you selected

let me try to upgrade to 2.10.18

ok, that worked.