Creates a CloudService object based on user settings.
Attributes:
| Name |
Type |
Description |
verbose |
bool
|
Whether to print verbose output.
|
cloud_settings |
dict
|
Cloud settings dictionary
|
Returns:
| Type |
Description |
Initialized CloudService object
|
|
Source code in openavmkit/cloud/cloud.py
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
71
72
73
74
75
76
77
78
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 | def init(
verbose: bool,
cloud_settings: dict
) -> CloudService | None:
"""Creates a CloudService object based on user settings.
Attributes
----------
verbose : bool
Whether to print verbose output.
cloud_settings : dict
Cloud settings dictionary
Returns
-------
Initialized CloudService object
"""
enabled = cloud_settings.get("enabled", True)
if not enabled:
print("Cloud service disabled, skipping...")
return None
if cloud_settings is None:
print("No cloud settings found, cannot initialize cloud...")
return None
cloud_type = cloud_settings.get("type", None)
if cloud_type is None:
print("Missing 'type' in cloud settings, cannot initialize cloud...")
return
cloud_type = cloud_type.lower()
cloud_access = _get_cloud_access(cloud_type)
if cloud_access is not None:
cloud_access = cloud_access.lower()
if cloud_settings is not None:
illegal_values = [
"hf_token",
"azure_storage_connection_string",
"sftp_password",
"sftp_username",
]
for key in illegal_values:
if key.lower() in cloud_settings:
raise ValueError(
f"Sensitive credentials '{key}' should never be stored in your settings file! They should ONLY be in your local .env file!"
)
warn_values = [
"azure_access"
"hf_access",
"sftp_access",
]
for key in warn_values:
if key.lower() in cloud_settings:
warnings.warn(f"Field '{key}' should be stored in your .env file, not in your cloud.json. The version in cloud.json will be ignored.")
anonymous_access_allowed = False
if cloud_type == "azure":
anonymous_access_allowed = True
# Add other conditions for anonymous access being allowed here
if anonymous_access_allowed:
if cloud_access is None:
cloud_access = "read_only"
if verbose:
print(f"Cloud access was None, defaulting to 'read_only'...")
elif cloud_access == "read_write":
anonymous_access_allowed = False
if verbose:
print(
f"Initializing cloud service of type '{cloud_type}' with access '{cloud_access}'..."
)
if cloud_access is None:
cloud_access_key = _get_cloud_access_key(cloud_type)
raise ValueError(
f"Missing '{cloud_access_key}' in environment. Have you created your .env file and properly filled it out?"
)
if anonymous_access_allowed:
try:
credentials = _get_creds_from_env(cloud_type)
except ValueError as e:
credentials = None
else:
credentials = _get_creds_from_env(cloud_type)
try:
cloud_service = _init_service(cloud_type, cloud_access, credentials, cloud_settings)
except ValueError as e:
print(f"Error initializing cloud! Error = {e}")
return None
cloud_service.verbose = verbose
return cloud_service
|