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
80 changes: 68 additions & 12 deletions artifactory.py
Original file line number Diff line number Diff line change
Expand Up @@ -1520,7 +1520,11 @@ class ArtifactoryOpensourceAccessor(_ArtifactoryAccessor):
# strings, and the add_slash() will literally append a slash character to the string
# path. See the original code in
# https://github.com/python/cpython/blob/v3.13.2/Lib/glob.py#L448-L510
class _ArtifactoryGlobber(glob._Globber if IS_PYTHON_3_13_OR_NEWER else object):
# In Python 3.14+, glob._Globber was removed, so we need to check if it exists
_HAS_GLOB_GLOBBER = IS_PYTHON_3_13_OR_NEWER and hasattr(glob, "_Globber")


class _ArtifactoryGlobber(glob._Globber if _HAS_GLOB_GLOBBER else object):
def recursive_selector(self, part, parts):
"""Returns a function that selects a given path and all its children,
recursively, filtering by pattern.
Expand Down Expand Up @@ -1599,7 +1603,7 @@ class PureArtifactoryPath(pathlib.PurePath):
# In Python 3.13, this attribute is accessed by PurePath.glob(), and we need to
# override it to behave properly for ArtifactoryPaths with a custom subclass of
# glob._Globber.
if IS_PYTHON_3_13_OR_NEWER:
if _HAS_GLOB_GLOBBER:
_globber = _ArtifactoryGlobber

__slots__ = ()
Expand Down Expand Up @@ -1713,14 +1717,31 @@ def __init__(self, *args, **kwargs):
if not IS_PYTHON_3_12_OR_NEWER:
return

# Extract custom kwargs that are not part of pathlib.Path.__init__
# These are ArtifactoryPath-specific parameters
custom_kwargs = {}
artifactory_params = {
"apikey",
"token",
"auth",
"auth_type",
"cert",
"session",
"timeout",
"verify",
}
for key in artifactory_params:
if key in kwargs:
custom_kwargs[key] = kwargs.pop(key)

super().__init__(*args, **kwargs)

cfg_entry = get_global_config_entry(self.drive)

# Auth section
apikey = kwargs.get("apikey")
token = kwargs.get("token")
auth_type = kwargs.get("auth_type")
apikey = custom_kwargs.get("apikey")
token = custom_kwargs.get("token")
auth_type = custom_kwargs.get("auth_type")

if apikey:
logger.debug("Use XJFrogApiAuth apikey")
Expand All @@ -1729,22 +1750,22 @@ def __init__(self, *args, **kwargs):
logger.debug("Use XJFrogArtBearerAuth token")
self.auth = XJFrogArtBearerAuth(token=token)
else:
auth = kwargs.get("auth")
auth = custom_kwargs.get("auth")
self.auth = auth if auth_type is None else auth_type(*auth)

if self.auth is None and cfg_entry:
auth = (cfg_entry["username"], cfg_entry["password"])
self.auth = auth if auth_type is None else auth_type(*auth)

self.cert = kwargs.get("cert")
self.session = kwargs.get("session")
self.timeout = kwargs.get("timeout")
self.cert = custom_kwargs.get("cert")
self.session = custom_kwargs.get("session")
self.timeout = custom_kwargs.get("timeout")

if self.cert is None and cfg_entry:
self.cert = cfg_entry["cert"]

if "verify" in kwargs:
self.verify = kwargs.get("verify")
if "verify" in custom_kwargs:
self.verify = custom_kwargs.get("verify")
elif cfg_entry:
self.verify = cfg_entry["verify"]
else:
Expand Down Expand Up @@ -1896,12 +1917,47 @@ def _scandir(self):
return self._accessor.scandir(self)

def glob(self, *args, **kwargs):
if IS_PYTHON_3_13_OR_NEWER:
if _HAS_GLOB_GLOBBER:
# In Python 3.13, the implementation of Path.glob() changed such that it assumes that it
# works only with real filesystem paths and will try to call real filesystem operations like
# os.scandir(). In Python 3.13, we explicitly intercept this and call PathBase's glob()
# implementation, which only depends on methods defined on the Path subclass.
return pathlib._abc.PathBase.glob(self, *args, **kwargs)
elif IS_PYTHON_3_12_OR_NEWER:
# In Python 3.14+, glob._Globber was removed but we still need custom glob behavior
# that doesn't rely on filesystem operations. We'll use a simplified implementation
# based on iterdir() and fnmatch that works with Artifactory's REST API.
pattern = args[0] if args else kwargs.get("pattern", "*")
parts = pattern.split("/")

def _glob_select(pat_parts, path):
if not pat_parts:
yield path
return
part = pat_parts[0]
rest = pat_parts[1:]

if part == "**":
yield from _glob_select(rest, path)
try:
for child in path.iterdir():
if child.is_dir():
yield from _glob_select(pat_parts, child)
except (OSError, PermissionError):
pass
else:
try:
for child in path.iterdir():
if fnmatch.fnmatch(child.name, part):
if rest:
if child.is_dir():
yield from _glob_select(rest, child)
else:
yield child
except (OSError, PermissionError):
pass

return _glob_select(parts, self)
return super().glob(*args, **kwargs)

def download_stats(self, pathobj=None):
Expand Down
1 change: 1 addition & 0 deletions setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Programming Language :: Python :: 3.14",
"Topic :: Software Development :: Libraries",
"Topic :: System :: Filesystems",
],
Expand Down
95 changes: 95 additions & 0 deletions tests/unit/test_artifactory_path.py
Original file line number Diff line number Diff line change
Expand Up @@ -1434,6 +1434,101 @@ def test_glob(self):
],
)

@responses.activate
def test_glob_recursive_pattern(self):
"""
Test that recursive glob with a pattern like **/*.gz works.
This is the default code example from the README.
"""
com_dir_stat = {
"repo": "libs-release-local",
"path": "/com",
"created": "2014-02-18T15:35:29.361+04:00",
"lastModified": "2014-02-18T15:35:29.361+04:00",
"lastUpdated": "2014-02-18T15:35:29.361+04:00",
"children": [
{"uri": "/foo.gz"},
{"uri": "/bar.txt"},
],
"uri": "http://artifactory.local/artifactory/api/storage/libs-release-local/com",
}
index_dir_stat = {
"repo": "libs-release-local",
"path": "/.index",
"created": "2014-02-18T15:35:29.361+04:00",
"lastModified": "2014-02-18T15:35:29.361+04:00",
"lastUpdated": "2014-02-18T15:35:29.361+04:00",
"children": [],
"uri": "http://artifactory.local/artifactory/api/storage/libs-release-local/.index",
}
ArtifactoryPath = self.cls
root_path = ArtifactoryPath(
"http://artifactory.local/artifactory/libs-release-local"
)
constructed_url = (
"http://artifactory.local/artifactory/api/storage/libs-release-local"
)

responses.add(responses.GET, constructed_url, status=200, json=self.dir_stat)
responses.add(responses.GET, constructed_url, status=200, json=self.dir_stat)
responses.add(
responses.GET,
f"{constructed_url}/.index",
status=200,
json=index_dir_stat,
)
responses.add(
responses.GET,
f"{constructed_url}/.index",
status=200,
json=index_dir_stat,
)
responses.add(
responses.GET,
f"{constructed_url}/.index",
status=200,
json=index_dir_stat,
)
responses.add(
responses.GET,
f"{constructed_url}/com",
status=200,
json=com_dir_stat,
)
responses.add(
responses.GET,
f"{constructed_url}/com",
status=200,
json=com_dir_stat,
)
responses.add(
responses.GET,
f"{constructed_url}/com",
status=200,
json=com_dir_stat,
)
responses.add(
responses.GET,
f"{constructed_url}/com/foo.gz",
status=200,
json=self.file_stat,
)
responses.add(
responses.GET,
f"{constructed_url}/com/bar.txt",
status=200,
json=self.file_stat,
)

results = list(root_path.glob("**/*.gz"))

self.assertEqual(
[str(r) for r in results],
[
"http://artifactory.local/artifactory/libs-release-local/com/foo.gz",
],
)


class ArtifactorySaaSPathTest(unittest.TestCase):
cls = artifactory.ArtifactorySaaSPath
Expand Down
1 change: 1 addition & 0 deletions tox.ini
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ envlist =
py311
py312
py313
py314
pre-commit

[testenv]
Expand Down