Skip to content

Commit

Permalink
apply code suggestions
Browse files Browse the repository at this point in the history
- make __find_webdav_storage as more reusable function
- use provider option for set the personal and shared modes instead of creating an option for it
- clean up code
  • Loading branch information
andre-code committed Nov 25, 2024
1 parent dc7ce49 commit f6cef33
Show file tree
Hide file tree
Showing 2 changed files with 23 additions and 39 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -227,11 +227,11 @@ def config_string(self, name: str) -> str:

if access == "shared" and storage_type == "polybox":
self.configuration["url"] = "https://polybox.ethz.ch/public.php/webdav/"
if access == "shared" and storage_type == "switchDrive":
elif access == "shared" and storage_type == "switchDrive":
self.configuration["url"] = "https://drive.switch.ch/public.php/webdav/"
if access == "personal" and storage_type == "polybox":
elif access == "personal" and storage_type == "polybox":
self.configuration["url"] = "https://polybox.ethz.ch/remote.php/webdav/"
if access == "personal" and storage_type == "switchDrive":
elif access == "personal" and storage_type == "switchDrive":
self.configuration["url"] = "https://drive.switch.ch/remote.php/webdav/"

# Extract the user from the public link
Expand Down
56 changes: 20 additions & 36 deletions components/renku_data_services/storage/rclone.py
Original file line number Diff line number Diff line change
Expand Up @@ -94,7 +94,7 @@ def __patch_schema_s3_endpoint_required(spec: list[dict[str, Any]]) -> None:
@staticmethod
def __patch_schema_add_switch_provider(spec: list[dict[str, Any]]) -> None:
"""Adds a fake provider to help with setting up switch storage."""
s3 = next(s for s in spec if s["Prefix"] == "s3")
s3 = RCloneValidator.__find_storage(spec, "s3")
providers = next(o for o in s3["Options"] if o["Name"] == "provider")
providers["Examples"].append({"Value": "Switch", "Help": "Switch Object Storage", "Provider": ""})
s3["Options"].append(
Expand Down Expand Up @@ -158,12 +158,12 @@ def __patch_schema_remove_oauth_propeties(spec: list[dict[str, Any]]) -> None:
storage["Options"] = options

@staticmethod
def __find_webdav_storage(spec: list[dict[str, Any]]) -> dict[str, Any]:
def __find_storage(spec: list[dict[str, Any]], prefix: str) -> dict[str, Any]:
"""Find and return the WebDAV storage schema from the spec."""
webdav_storage = next((s for s in spec if s["Prefix"] == "webdav"), None)
if not webdav_storage:
raise errors.ValidationError(message="WebDAV storage not found in schema.")
return deepcopy(webdav_storage)
storage = next((s for s in spec if s["Prefix"] == prefix), None)
if not storage:
raise errors.ValidationError(message=f"'{prefix}' storage not found in schema.")
return deepcopy(storage)

@staticmethod
def __add_webdav_based_storage(
Expand All @@ -176,36 +176,10 @@ def __add_webdav_based_storage(
) -> None:
"""Create a modified copy of WebDAV storage and add it to the schema."""
# Find WebDAV storage schema and create a modified copy
storage_copy = RCloneValidator.__find_webdav_storage(spec)
storage_copy = RCloneValidator.__find_storage(spec, "webDav")
storage_copy.update({"Prefix": prefix, "Name": name, "Description": description})

custom_options = [
{
"Name": "access",
"Help": "Choose the mode to access the data source.",
"Provider": "",
"Default": "",
"Value": None,
"Examples": [
{"Value": "personal", "Help": "Use Private to connect a folder that only you use", "Provider": ""},
{
"Value": "shared",
"Help": "To connect a folder you share with others, both personal & shared folders.",
"Provider": "",
},
],
"Required": False,
"Type": "string",
"ShortOpt": "",
"Hide": 0,
"IsPassword": False,
"NoPrefix": False,
"Advanced": False,
"Exclusive": False,
"Sensitive": False,
"DefaultStr": "",
"ValueStr": "",
},
custom_option = [
{
"Name": "public_link",
"Help": public_link_help,
Expand All @@ -226,11 +200,21 @@ def __add_webdav_based_storage(
"Type": "string",
},
]
storage_copy["Options"].extend(custom_options)
storage_copy["Options"].extend(custom_option)

# use provider to indicate if the option is for an personal o shared storage
for option in storage_copy["Options"]:
if option["Name"] == "url":
if option["Name"] == "provider":
option["example"] = [
{"Value": "personal", "Help": "Use Private to connect a folder that only you use", "Provider": ""},
{
"Value": "shared",
"Help": "To connect a folder you share with others, both personal & shared folders.",
"Provider": "",
},
]
option["Exclusive"] = True
elif option["Name"] == "url":
option.update({"Provider": "personal", "Default": url_value, "Required": False})
elif option["Name"] in ["bearer_token", "bearer_token_command", "headers", "user"]:
option["Provider"] = "personal"
Expand Down

0 comments on commit f6cef33

Please sign in to comment.