Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 13 additions & 5 deletions databusclient/api/deploy.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@

import requests

from databusclient.api.utils import validate_databus_version_uri

_debug = False


Expand Down Expand Up @@ -352,13 +354,19 @@ def create_dataset(
OPTIONAL! Metadata for the Group: Description. NOTE: Is only used if all group metadata is set
"""

_versionId = str(version_id).strip("/")
parts = _versionId.rsplit("/", 4)
if len(parts) < 5:
if len(artifact_version_abstract.strip()) > 200:
raise BadArgumentException(
f"Invalid version_id format: '{version_id}'. "
f"Expected format: <BASE>/<ACCOUNT>/<GROUP>/<ARTIFACT>/<VERSION>"
f"Artifact & version abstract exceeds maximum allowed length of 200 characters "
f"(got {len(artifact_version_abstract.strip())})."
)

try:
validate_databus_version_uri(version_id)
except ValueError as e:
raise BadArgumentException(str(e)) from e

_versionId = str(version_id).strip("/")
parts = _versionId.rsplit("/", 4)
_, _account_name, _group_name, _artifact_name, version = parts

# could be build from stuff above,
Expand Down
52 changes: 51 additions & 1 deletion databusclient/api/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,18 @@
`download`, `deploy` and `delete` modules.
"""

from typing import Optional, Tuple
import hashlib
import re
from typing import Optional, Tuple
from urllib.parse import urlparse
import requests

# Regex for Databus identifier components (account, group, artifact, version)
_DATABUS_ID_RE = re.compile(r"^[a-zA-Z0-9_.-]+$")
# Regex for authority (host:port)
_DATABUS_AUTHORITY_RE = re.compile(r"^[a-zA-Z0-9.-]+(?::[0-9]+)?$")



def get_databus_id_parts_from_file_url(
uri: str,
Expand Down Expand Up @@ -71,3 +79,45 @@ def compute_sha256_and_length(filepath):
sha256.update(chunk)
total_length += len(chunk)
return sha256.hexdigest(), total_length


def validate_databus_version_uri(uri: str) -> None:
"""Validate a Databus version URI format.

Expects format: http(s)://<AUTHORITY>/<ACCOUNT>/<GROUP>/<ARTIFACT>/<VERSION>

Raises:
ValueError: If URI scheme, authority, trailing slashes, or path components are invalid.
"""
if not uri or not isinstance(uri, str):
raise ValueError("Databus version_id must be a non-empty string.")

if not (uri.startswith("http://") or uri.startswith("https://")):
raise ValueError(
f"Invalid version_id URI scheme: '{uri}'. Must start with 'http://' or 'https://'."
)

parsed = urlparse(uri)
if not parsed.netloc or not _DATABUS_AUTHORITY_RE.match(parsed.netloc):
raise ValueError(
f"Invalid authority in version_id URI: '{parsed.netloc or uri}'."
)

# Preserve slash delimiters: path must start with '/' followed by account/group/artifact/version
# Any trailing slash, double slash, or extra segment creates empty or extra parts.
path_segments = parsed.path.split("/")[1:]

if len(path_segments) != 4:
raise ValueError(
f"Invalid version_id format: '{uri}'. Expected format: <BASE>/<ACCOUNT>/<GROUP>/<ARTIFACT>/<VERSION>"
)

component_names = ["ACCOUNT", "GROUP", "ARTIFACT", "VERSION"]
for name, seg in zip(component_names, path_segments):
if not seg or not _DATABUS_ID_RE.match(seg):
raise ValueError(
f"Invalid Databus {name} component '{seg}' in version_id: '{uri}'. "
"Must be non-empty and contain only alphanumeric characters, underscores, hyphens, or dots."
)


35 changes: 35 additions & 0 deletions tests/test_deploy.py
Original file line number Diff line number Diff line change
Expand Up @@ -128,3 +128,38 @@ def test_empty_cvs():
}

assert dataset == correct_dataset


def test_abstract_length_validation():
long_abstract = "a" * 205
with pytest.raises(BadArgumentException, match="exceeds maximum allowed length of 200 characters"):
create_dataset(
version_id="https://databus.dbpedia.org/user/group/artifact/1.0.0",
artifact_version_title="Test Title",
artifact_version_abstract=long_abstract,
artifact_version_description="Test description",
license_url="https://dalicc.net/licenses/cc-by-4.0",
distributions=[],
)


def test_version_id_uri_validation():
invalid_uris = [
"databus.dbpedia.org/user/group/artifact/1.0.0", # missing scheme
"https://databus.dbpedia.org/user/group/artifact", # missing version part
"https://databus.dbpedia.org/user//artifact/1.0.0", # empty group part
"https://databus.dbpedia.org/user/group/artifact/1.0.0/", # trailing slash
"https://databus.dbpedia.org/user/gr oup/artifact/1.0.0", # component containing space
]
for invalid_uri in invalid_uris:
with pytest.raises(BadArgumentException):
create_dataset(
version_id=invalid_uri,
artifact_version_title="Test Title",
artifact_version_abstract="Valid abstract",
artifact_version_description="Test description",
license_url="https://dalicc.net/licenses/cc-by-4.0",
distributions=[],
)