diff --git a/databusclient/api/deploy.py b/databusclient/api/deploy.py
index 08d9016..1fd93c7 100644
--- a/databusclient/api/deploy.py
+++ b/databusclient/api/deploy.py
@@ -12,6 +12,8 @@
import requests
+from databusclient.api.utils import validate_databus_version_uri
+
_debug = False
@@ -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: ////"
+ 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,
diff --git a/databusclient/api/utils.py b/databusclient/api/utils.py
index df4f49d..82ae470 100644
--- a/databusclient/api/utils.py
+++ b/databusclient/api/utils.py
@@ -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,
@@ -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)://////
+
+ 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: ////"
+ )
+
+ 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."
+ )
+
+
diff --git a/tests/test_deploy.py b/tests/test_deploy.py
index 679f11f..95f6863 100644
--- a/tests/test_deploy.py
+++ b/tests/test_deploy.py
@@ -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=[],
+ )
+
+