1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
|
# ---------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# ---------------------------------------------------------
# pylint: disable=protected-access
from typing import Any, Optional, Union, cast
from azure.ai.ml._restclient.v2023_04_01_preview import models
from azure.ai.ml._restclient.v2024_07_01_preview import models as models2024
from azure.ai.ml.entities._credentials import (
AccountKeyConfiguration,
CertificateConfiguration,
NoneCredentialConfiguration,
SasTokenConfiguration,
ServicePrincipalConfiguration,
)
from azure.ai.ml.entities._datastore._on_prem_credentials import KerberosKeytabCredentials, KerberosPasswordCredentials
def from_rest_datastore_credentials(
rest_credentials: models.DatastoreCredentials,
) -> Union[
AccountKeyConfiguration,
SasTokenConfiguration,
ServicePrincipalConfiguration,
CertificateConfiguration,
NoneCredentialConfiguration,
]:
config_class: Any = NoneCredentialConfiguration
if isinstance(rest_credentials, (models.AccountKeyDatastoreCredentials, models2024.AccountKeyDatastoreCredentials)):
# we are no more using key for key base account.
# https://github.com/Azure/azure-sdk-for-python/pull/35716
if isinstance(rest_credentials.secrets, models2024.SasDatastoreSecrets):
config_class = SasTokenConfiguration
else:
config_class = AccountKeyConfiguration
elif isinstance(rest_credentials, (models.SasDatastoreCredentials, models2024.SasDatastoreCredentials)):
config_class = SasTokenConfiguration
elif isinstance(
rest_credentials, (models.ServicePrincipalDatastoreCredentials, models2024.ServicePrincipalDatastoreCredentials)
):
config_class = ServicePrincipalConfiguration
elif isinstance(
rest_credentials, (models.CertificateDatastoreCredentials, models2024.CertificateDatastoreCredentials)
):
config_class = CertificateConfiguration
return cast(
Union[
AccountKeyConfiguration,
SasTokenConfiguration,
ServicePrincipalConfiguration,
CertificateConfiguration,
NoneCredentialConfiguration,
],
config_class._from_datastore_rest_object(rest_credentials),
)
def _from_rest_datastore_credentials_preview(
rest_credentials: models.DatastoreCredentials,
) -> Optional[Union[KerberosKeytabCredentials, KerberosPasswordCredentials]]:
if isinstance(rest_credentials, models.KerberosKeytabCredentials):
return KerberosKeytabCredentials._from_rest_object(rest_credentials)
if isinstance(rest_credentials, models.KerberosPasswordCredentials):
return KerberosPasswordCredentials._from_rest_object(rest_credentials)
return None
|
