Skip to content

Credentials

Credential classes used to perform authenticated interactions with Azure

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
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
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."
        ),
    )

    @root_validator
    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 bool(has_account_url ^ 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)

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

Source code in prefect_azure/credentials.py
122
123
124
125
126
127
128
129
130
131
132
133
134
135
@root_validator
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 bool(has_account_url ^ 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
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
@_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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
@_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
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
@_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
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
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`."
        ),
    )

    @root_validator
    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
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
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
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
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)

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
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
@root_validator
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

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
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
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
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
@_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
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
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
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
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
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
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
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
@_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