"""
cMeta repositories manager
cMeta author and developer: (C) 2025-2026 Grigori Fursin
See the cMeta COPYRIGHT and LICENSE files in the project root for details.
"""
import json
import logging
from typing import Dict, Any, List, Optional
from pathlib import Path
import os
import sys
import fnmatch
import time
from . import utils
from .utils.common import _error
# Migration stubs followed in this process, so that each notice is printed once
_migrated_notices = set()
# How many stubs in a row a lookup follows before it gives up (a stub pointing to a stub)
MAX_MIGRATED_HOPS = 8
# The index lock (`<home>/index.lock`): a full reindex holds it for its whole rebuild and swap, the index of
# pulled or plugged repositories for its duration, every write of a record for a moment. A writer that finds
# it held waits (saying so after INDEX_LOCK_NOTICE_SECONDS) up to INDEX_LOCK_TIMEOUT seconds, which the
# environment variable changes.
INDEX_LOCK_TIMEOUT_ENV = 'CMETA_INDEX_LOCK_TIMEOUT'
INDEX_LOCK_TIMEOUT = 600
INDEX_LOCK_NOTICE_SECONDS = 3
[docs]
class Repos:
"""
cMeta repositories manager.
"""
###################################################################################################
[docs]
def __init__(
self,
cfg: Dict[str, Any], # Configuration dictionary.
home_path: Path, # Path to repositories directory.
index_path: Path, # Path to index directory.
repos_config_path: Path, # Path to repositories config file.
logger: logging.Logger = None, # Logger instance (optional).
index_extension: str = '.pkl', # Index file extension.
fail_on_error: bool = False, # If True, raise exception on error.
match_version_func = None, # Callback used to evaluate version matching rules.
):
"""
Initialize Repos manager.
Args:
cfg (Dict[str, Any]): Configuration dictionary.
home_path (Path): Path to repositories directory.
index_path (Path): Path to index directory.
repos_config_path (Path): Path to repositories config file.
logger (logging.Logger | None): Logger instance (optional).
index_extension (str): Index file extension.
fail_on_error (bool): If True, raise exception on error.
match_version_func: Callback used to evaluate version matching rules.
Returns:
dict: Operation result.
Raises:
Exception: Propagated runtime errors, if any.
"""
self.cfg = cfg
self.home_path = home_path
self.index_path = index_path
self.file_cache = {}
self.index_extension = index_extension
self.repos_config_path = repos_config_path
self._repositories = []
self.fail_on_error = fail_on_error
self.match_version_func = match_version_func
# Create a child logger that inherits CMeta's configuration
self.logger = logging.getLogger(__name__) if logger is None else logger.getChild("repos")
self.logger.debug("Initializing Repos class ...")
self.KEY_INDEX_UIDS = 'uids'
self.KEY_INDEX_LOWERCASE_ALIASES = 'lowercase_aliases'
# In repos.json: a registered repository whose folder could not be reached when the index was last built
# (a detached drive, a share that is down). It stays registered and is indexed again when it is back.
self.KEY_REPO_UNREACHABLE = 'unreachable'
self._soft_lock_registered = False # the repositories that ask for the soft lock were read once (init)
###################################################################################################
[docs]
def init(
self,
con: bool = False, # Console mode flag (for compatibility).
verbose: bool = False, # Enable verbose output (for compatibility).
):
"""
Initialize repositories and index if running for the first time.
Creates repos directory, local repository, and triggers reindexing if needed.
Args:
con (bool): Console mode flag (for compatibility).
verbose (bool): Enable verbose output (for compatibility).
Returns:
dict: Dictionary with 'return': 0 on success, or 'return' > 0 and 'error' on failure.
Raises:
Exception: Propagated runtime errors, if any.
"""
trigger_reindex = False
home_path_local = os.path.join(self.home_path, 'local')
# Check if repos do not exist (exist_ok: several first runs on one home at once)
if not os.path.isdir(self.home_path):
self.logger.debug(f"Creating repos directory in {self.home_path} ...")
os.makedirs(self.home_path, exist_ok=True)
# Check local repo there
if not os.path.isdir(home_path_local):
self.logger.debug(f"Creating local repo directory in {home_path_local} ...")
os.makedirs(home_path_local, exist_ok=True)
repo_local_meta_file = os.path.join(home_path_local, self.cfg['repo_meta_desc'])
if not os.path.isfile(repo_local_meta_file):
repo_local_meta = self.cfg['repo_local_meta'].copy()
repo_local_meta['category'] = 'repo,' + self.cfg['category_repo_uid']
r = utils.files.safe_write_file(repo_local_meta_file, repo_local_meta, fail_on_error=self.fail_on_error, logger=self.logger)
if r['return']>0: return r
# Check if repo file exists
if not os.path.isfile(self.repos_config_path):
trigger_reindex = True
# Need ordered dict (Python >= 3.7)
repos_paths = {}
# First local
repos_paths[home_path_local]={}
# Then internal repo
this_module_path = os.path.dirname(os.path.abspath(__file__))
internal_repo_path = os.path.join(this_module_path, 'internal-repo')
repos_paths[internal_repo_path]={}
# Do not sort keys!
r = utils.files.safe_write_file(self.repos_config_path, repos_paths, fail_on_error=self.fail_on_error, logger=self.logger, sort_keys=False)
if r['return']>0: return r
if trigger_reindex or not os.path.isdir(self.index_path):
# Under the index lock: a process that arrives while another one builds or swaps the index
# waits for it and then finds the index there, instead of building its own (several first
# runs on one home at once, or a start during the swap of a full reindex)
r = self._lock_index('the first run')
if r['return']>0: return r
index_lock = r['lock']
try:
if trigger_reindex or not os.path.isdir(self.index_path):
r = self._index_full(con=con, verbose=verbose)
if r['return'] >0: return r
finally:
index_lock.release()
# The repositories that ask for the soft lock (`lock: soft` in their descriptor) are known to the path
# lock of this process from the start (once per instance: init runs with every access)
if not self._soft_lock_registered:
r = self._register_soft_lock_repos()
if r['return']>0: return r
self._soft_lock_registered = True
return {'return':0}
###################################################################################################
def _register_soft_lock(
self,
path, # The registered folder of the repository.
repo_meta, # Its descriptor (_cmr.yaml, with the extra meta of the registration).
):
"""
A repository whose descriptor says `lock: soft` - one that several systems write at once through
a shared drive (a Windows tree plugged from WSL2 or a container, a network share), where the OS
locks of the systems do not see each other - asks the path lock for the soft lock under its folder
(`utils.files.use_soft_lock_under`), for this process.
"""
if path and str(repo_meta.get('lock', '')).strip().lower() == 'soft':
utils.files.use_soft_lock_under(path)
self.logger.debug(f"repository {path} asks for the soft lock (lock: soft)")
def _register_soft_lock_repos(self):
"""
At the start of a process: the repositories of the index that ask for the soft lock (see
`_register_soft_lock`). The index of repositories is small and read through the cache.
"""
index_repo_file = os.path.join(self.index_path, 'repo' + self.index_extension)
if not os.path.isfile(index_repo_file):
return {'return':0}
r = utils.files.safe_read_file_via_cache(index_repo_file, self.file_cache, fail_on_error=self.fail_on_error, logger=self.logger)
if r['return']>0: return r
for entry in (r['data'].get(self.KEY_INDEX_UIDS) or {}).values():
self._register_soft_lock(entry.get('path'), entry.get('_cmr') or entry.get('cmeta') or {})
return {'return':0}
###################################################################################################
[docs]
def add_to_index(
self,
cmeta, # Artifact metadata dictionary.
cmeta_ref_parts, # Parsed cMeta reference components.
path, # Filesystem path.
original_alias = None, # Previous artifact alias before update.
original_uid = None, # Previous artifact UID before update.
sharding_slices_num = 0, # Number of sharding slices
clean_aliases = False, # If True, drop every other alias of this UID from the alias index (a folder renamed by hand), in the same locked write.
index_lock = None, # The index lock when the caller holds it already (lock_index): the record is written under it, without taking it again.
):
"""
Insert or update an artifact record in the category index (under the index lock, see `_lock_index`).
Args:
cmeta: Artifact metadata dictionary.
cmeta_ref_parts: Parsed cMeta reference components.
path: Filesystem path.
original_alias: Previous artifact alias before update.
original_uid: Previous artifact UID before update.
sharding_slices_num (int): Number of UID slices used to shard artifact paths.
clean_aliases (bool): If True, every alias of the alias index that still lists this UID
and is not the alias of the record is dropped (the stale aliases a folder renamed by
hand leaves behind), in the same locked write. Aliases of other UIDs are not touched.
index_lock: The index lock when the caller holds it already (`lock_index`): the record is
written under it, without taking it again (the lock is not re-entrant).
Returns:
dict: Operation result, with 'dropped_aliases': the (lowercase) aliases dropped by `clean_aliases`.
Raises:
Exception: Propagated runtime errors, if any.
"""
# Under the index lock: never into an index that a running full reindex is about to replace
if index_lock is not None:
return self._add_to_index(cmeta, cmeta_ref_parts, path, original_alias, original_uid, sharding_slices_num, clean_aliases)
r = self._lock_index('the write of a record')
if r['return']>0: return r
index_lock = r['lock']
try:
return self._add_to_index(cmeta, cmeta_ref_parts, path, original_alias, original_uid, sharding_slices_num, clean_aliases)
finally:
index_lock.release()
def _add_to_index(self, cmeta, cmeta_ref_parts, path, original_alias, original_uid, sharding_slices_num, clean_aliases):
"""`add_to_index` with the index lock held by the caller."""
category_alias = cmeta_ref_parts['category_alias'].lower()
artifact_uid = cmeta_ref_parts['artifact_uid']
artifact_alias = cmeta_ref_parts.get('artifact_alias')
# Data is taken directly from cache to be fast - it should not be changed externally
index_file = os.path.join(self.index_path, category_alias + self.index_extension)
r = utils.files.safe_read_file(index_file, lock=True, keep_locked=True, fail_on_error=self.fail_on_error, logger=self.logger)
if r['return']>0:
if r['return']!=16: return r
index_data = {}
index_file_lock = None
else:
index_data = r['data']
index_file_lock = r['file_lock']
uids = index_data.setdefault(self.KEY_INDEX_UIDS, {})
lowercase_aliases = index_data.setdefault(self.KEY_INDEX_LOWERCASE_ALIASES, {})
# If needed, delete the original one before adding the new/updated one
if original_alias is not None:
artifact_alias_lowercase = original_alias.lower()
lowercase_alias_uids = lowercase_aliases.get(artifact_alias_lowercase, [])
if original_uid in lowercase_alias_uids:
lowercase_alias_uids.remove(original_uid)
if len(lowercase_alias_uids) == 0:
del(lowercase_aliases[artifact_alias_lowercase])
if original_uid is not None and original_uid in uids:
del(uids[original_uid])
# Prepare new record
record = {'cmeta': cmeta, 'cmeta_ref_parts': cmeta_ref_parts, 'path': path}
if sharding_slices_num is not None and sharding_slices_num != 0:
record['sharding_slices_num'] = sharding_slices_num
uids[artifact_uid] = record
artifact_alias_lowercase = None
if artifact_alias is not None:
artifact_alias_lowercase = artifact_alias.lower()
lowercase_alias_uids = lowercase_aliases.get(artifact_alias_lowercase, [])
if artifact_uid not in lowercase_alias_uids:
lowercase_alias_uids.append(artifact_uid)
lowercase_aliases[artifact_alias_lowercase] = lowercase_alias_uids
# The other aliases that still point at this UID (a folder renamed by hand keeps its old
# alias in the index): only this UID is removed from them, other UIDs under the same alias stay
dropped_aliases = []
if clean_aliases:
for lowercase_alias in list(lowercase_aliases.keys()):
if lowercase_alias == artifact_alias_lowercase:
continue
lowercase_alias_uids = lowercase_aliases[lowercase_alias]
if artifact_uid in lowercase_alias_uids:
lowercase_alias_uids.remove(artifact_uid)
dropped_aliases.append(lowercase_alias)
if len(lowercase_alias_uids) == 0:
del(lowercase_aliases[lowercase_alias])
# Use atomic write to avoid corrupting large index files
r = utils.files.safe_write_file(index_file, index_data, file_lock=index_file_lock, atomic=True, fail_on_error=self.fail_on_error, logger=self.logger, sort_keys=False)
if r['return']>0: return r
# r = utils.files.safe_write_file(os.path.splitext(index_file)[0] + ".json", index_data, atomic=True, fail_on_error=self.fail_on_error, logger=self.logger)
# if r['return']>0: return r
return {'return':0, 'dropped_aliases': dropped_aliases}
###################################################################################################
[docs]
def remove_from_index(
self,
index_file, # Path to index file.
artifact_uid, # UID of artifact to remove.
artifact_alias_lowercase, # Lowercase alias of artifact to remove from alias index.
):
"""
Remove an artifact from the repository index.
Args:
index_file: Path to index file.
artifact_uid: UID of artifact to remove.
artifact_alias_lowercase: Lowercase alias of artifact to remove from alias index.
Returns:
dict: Dictionary with 'return': 0 on success, or 'return' > 0 and 'error' on failure.
Raises:
Exception: Propagated runtime errors, if any.
"""
r = self._lock_index('the removal of a record')
if r['return']>0: return r
index_lock = r['lock']
try:
return self._remove_from_index(index_file, artifact_uid, artifact_alias_lowercase)
finally:
index_lock.release()
def _remove_from_index(self, index_file, artifact_uid, artifact_alias_lowercase):
"""`remove_from_index` with the index lock held by the caller."""
r = utils.files.safe_read_file(index_file, lock=True, keep_locked=True, fail_on_error=self.fail_on_error, logger=self.logger)
if r['return']>0:
if r['return']!=16: return r
index_data = {}
index_file_lock = None
else:
index_data = r['data']
index_file_lock = r['file_lock']
if artifact_alias_lowercase is not None:
lowercase_aliases = index_data.setdefault(self.KEY_INDEX_LOWERCASE_ALIASES, {})
lowercase_alias_uids = lowercase_aliases.get(artifact_alias_lowercase, [])
if artifact_uid in lowercase_alias_uids:
lowercase_alias_uids.remove(artifact_uid)
if len(lowercase_alias_uids) == 0:
del(lowercase_aliases[artifact_alias_lowercase])
uids = index_data.setdefault(self.KEY_INDEX_UIDS, {})
if artifact_uid in uids:
del(uids[artifact_uid])
r = utils.files.safe_write_file(index_file, index_data, file_lock=index_file_lock, atomic=True, fail_on_error=self.fail_on_error, logger=self.logger)
if r['return']>0: return r
# r = utils.files.safe_write_file(os.path.splitext(index_file)[0] + ".json", index_data, atomic=True, fail_on_error=self.fail_on_error, logger=self.logger)
# if r['return']>0: return r
return {'return':0}
###################################################################################################
[docs]
def find_in_index(
self,
category_alias, # Lowercase category alias.
category_uid, # Category UID.
artifact_alias = None, # Artifact alias to search for.
artifact_uid = None, # Artifact UID to search for.
repos = [], # List of repository names to search in.
only_uids = False, # If True, return only UIDs without full metadata.
add_index_file = False, # If True, include index_file path in result.
skip_uids = False, # If True, skip UID validation.
):
"""
Find artifacts in the repository index.
Args:
category_alias: Lowercase category alias.
category_uid: Category UID.
artifact_alias: Artifact alias to search for.
artifact_uid: Artifact UID to search for.
repos: List of repository names to search in.
only_uids: If True, return only UIDs without full metadata.
add_index_file: If True, include index_file path in result.
skip_uids: If True, skip UID validation.
Returns:
dict: Dictionary with 'return': 0 and 'lst' containing found artifacts,
or 'return' > 0 and 'error' on failure.
Raises:
Exception: Propagated runtime errors, if any.
"""
category_alias = category_alias.lower()
# Data is taken directly from cache to be fast - it should not be changed externally
index_file = os.path.join(self.index_path, category_alias + self.index_extension)
r = utils.files.safe_read_file_via_cache(index_file, cache=self.file_cache, fail_on_error=self.fail_on_error, logger=self.logger)
if r['return']>0:
if r['return']!=16: return r
index = {}
else:
# Data from cache - do not change!
index = r['data']
artifact_uids = []
if artifact_uid is not None and artifact_uid != "":
if artifact_uid not in index.get(self.KEY_INDEX_UIDS, {}):
x = f'"{artifact_alias}" ({artifact_uid})' if artifact_alias is not None and artifact_alias != '' else f'{artifact_uid}'
err = f'{category_alias} artifact {x} not found'
return _error(err, 16, None, self.fail_on_error)
artifact_uids.append(artifact_uid)
elif artifact_alias is not None and artifact_alias != "":
artifact_alias_lowercase = artifact_alias.lower()
if '*' in artifact_alias or '?' in artifact_alias:
check_artifact_uids = list(index.get(self.KEY_INDEX_UIDS, {}).keys())
for artifact_uid in check_artifact_uids:
if artifact_uid not in index[self.KEY_INDEX_UIDS]:
return _error(f'corrupted {category_alias} UID "{artifact_uid}" not found in the index"', 1, None, self.fail_on_error)
record = index[self.KEY_INDEX_UIDS][artifact_uid]
lowercase_alias = record['cmeta_ref_parts'].get('artifact_alias_lowercase', None)
if lowercase_alias is None:
lowercase_alias = record['cmeta_ref_parts'].get('artifact_alias', '')
if not skip_uids and lowercase_alias == '':
lowercase_alias = artifact_uid
if fnmatch.fnmatch(lowercase_alias, artifact_alias_lowercase):
artifact_uids.append(artifact_uid)
else:
if artifact_alias_lowercase not in index.get(self.KEY_INDEX_LOWERCASE_ALIASES, {}):
# We should not be failing below even on debug to handle multiple-search - we need to handle aggregated search results
x_artifact_alias = "artifacts" if artifact_alias == '' or artifact_alias == None or artifact_alias == '*' else f'"{artifact_alias}"'
return _error(f'{category_alias} {x_artifact_alias} not found', 16, None, False) #self.fail_on_error)
artifact_uids.extend(index[self.KEY_INDEX_LOWERCASE_ALIASES][artifact_alias_lowercase])
else:
artifact_uids = list(index.get(self.KEY_INDEX_UIDS, {}).keys())
result = {'return':0, 'index_file': index_file, 'index': index}
# Check if prune by repos
if repos or category_uid is not None:
pruned_artifact_uids = []
for artifact_uid in artifact_uids:
if artifact_uid not in index[self.KEY_INDEX_UIDS]:
return _error(f'corrupted index for {category_alias} UID "{artifact_uid}"', 1, None, self.fail_on_error)
artifact_cmeta_ref_parts = index[self.KEY_INDEX_UIDS][artifact_uid]['cmeta_ref_parts']
# Note that "repo" artifacts are not inside any repo, i.e. their index entries
# have no "repo_uid" - they can never match a repo filter
if repos and artifact_cmeta_ref_parts.get('repo_uid') not in repos:
continue
if category_uid is not None and artifact_cmeta_ref_parts['category_uid'] != category_uid:
continue
pruned_artifact_uids.append(artifact_uid)
artifact_uids = pruned_artifact_uids
if only_uids:
result['artifact_uids'] = artifact_uids
else:
artifacts = []
for artifact_uid in artifact_uids:
if artifact_uid not in index[self.KEY_INDEX_UIDS]:
return _error(f'corrupted index for {category_alias} UID "{artifact_uid}"', 1, None, self.fail_on_error)
# A shallow copy of the record: its 'cmeta' is the cached object itself - do not change
# it (`find` copies it for the records it returns)
artifact = index[self.KEY_INDEX_UIDS][artifact_uid].copy()
if add_index_file:
artifact['index_file'] = index_file
artifacts.append(artifact)
result['artifacts'] = artifacts
return result
###################################################################################################
[docs]
def find_in_file_system(
self,
category_meta, # Category metadata dictionary used for file-system search.
category_alias, # Category alias (lowercase)
category_uid, # Category UID
artifact_alias = None, # Optional artifact alias (supports wildcards)
artifact_uid = None, # Optional artifact UID
repo_uids = [], # List of repository UIDs to limit file-system search.
):
"""
Find artifacts by scanning directories instead of using index.
Used for categories with no_index flag.
Args:
category_meta: Category metadata dictionary used for file-system search.
category_alias: Category alias (lowercase)
category_uid: Category UID
artifact_alias: Optional artifact alias (supports wildcards)
artifact_uid: Optional artifact UID
repo_uids: List of repository UIDs to limit file-system search.
Returns:
dict: {'return': 0, 'artifacts': [...]} or error
Raises:
Exception: Propagated runtime errors, if any.
"""
# Get repo artifacts
repo_artifacts = []
if repo_uids == None:
repo_uids = ['*']
for repo_uid in repo_uids:
repo_alias = None
if repo_uid == '*':
repo_alias = '*'
repo_uid = None
r = self.find_in_index('repo', self.cfg['category_repo_uid'], repo_alias, repo_uid)
if r['return'] >0: return r
repo_artifacts += r['artifacts']
# Iterate over repos:
artifacts = []
for repo_artifact in repo_artifacts:
repo_meta = repo_artifact['cmeta']
repo_path = repo_artifact['path']
# Take into account subdir if exists
full_repo_path = _get_full_path(repo_path, repo_meta)
repo_cmeta_ref_parts = repo_artifact['cmeta_ref_parts']
repo_alias = repo_cmeta_ref_parts.get('artifact_alias')
repo_uid = repo_cmeta_ref_parts['artifact_uid']
if category_uid in repo_meta.get('sharding_slices', {}):
sharding_slices = repo_meta['sharding_slices'][category_uid]
else:
sharding_slices = category_meta.get('sharding_slices')
# Get category path
path_to_category = os.path.join(full_repo_path, category_alias)
if os.path.isdir(path_to_category):
# Look for artifacts
r = self._find_artifacts(repo_meta, repo_alias, repo_uid, category_meta, category_alias, category_uid, path_to_category,
False, False, None, 0,
artifact_alias = artifact_alias, artifact_uid = artifact_uid)
if r['return'] >0: return r
for a in r['artifacts']:
if sharding_slices:
a['sharding_slices_num'] = len(sharding_slices)
artifacts.append(a)
return {'return': 0, 'artifacts': artifacts}
###################################################################################################
[docs]
def find(
self,
cmeta_ref: str, # cMeta reference string or parsed dictionary.
add_index_file: bool = False, # If True, include index_file path in result.
tags: str = None, # Optional tags to filter results.
skip_uids: bool = False, # If True, skip UID validation.
skip_non_indexed: bool = False, # Value for skip non indexed.
match: bool = None, # Optional match dictionary to filter results (key that ends with - is supported)
match_empty_version: bool = False, # Value for match empty version.
match_empty_values: bool = False,
all_tags: str = None, # match all tags in cmeta
follow_migrated: bool = True, # Follow a migration stub found by its alias alone to its artifact.
):
"""
Find artifacts by cMeta reference.
Args:
cmeta_ref (str): cMeta reference string or parsed dictionary.
add_index_file (bool): If True, include index_file path in result.
tags (str): Optional tags to filter results.
skip_uids (bool): If True, skip UID validation.
skip_non_indexed (bool): Value for skip non indexed.
match (bool): Optional match dictionary to filter results (key that ends with - is supported)
match_empty_version (bool): Value for match empty version.
match_empty_values (bool): Match when queried values or key components are empty.
all_tags (str): match all tags in cmeta
follow_migrated (bool): When an artifact is looked up by its alias alone (no UID, no
wildcards) and the alias is a migration stub (meta key `migrated_to`, see the
`migrate` command), return the artifact the stub points to instead, with a notice
on stdout. A lookup with a UID always returns exactly what it names.
Returns:
dict: Dictionary with 'return': 0 and 'artifacts' list on success,
or 'return' > 0 and 'error' on failure.
Raises:
Exception: Propagated runtime errors, if any.
"""
# Parse cMeta ref
if isinstance(cmeta_ref, str):
r = utils.names.parse_cmeta_ref(cmeta_ref, fail_on_error = self.fail_on_error)
if r['return'] >0: return r
cmeta_ref_parts = r['ref_parts']
else:
cmeta_ref_parts = cmeta_ref
# Check tags
if tags != None:
r = utils.common.normalize_tags(tags, fail_on_error = self.fail_on_error)
if r['return'] >0: return r
tags = r['tags']
# Check all tags
if all_tags != None:
r = utils.common.normalize_tags(all_tags, fail_on_error = self.fail_on_error)
if r['return'] >0: return r
all_tags = r['tags']
# Unpack to search
category_alias = cmeta_ref_parts.get('category_alias')
category_uid = cmeta_ref_parts.get('category_uid')
category_repo_alias = cmeta_ref_parts.get('category_repo_alias')
category_repo_uid = cmeta_ref_parts.get('category_repo_uid')
artifact_alias = cmeta_ref_parts.get('artifact_alias')
artifact_uid = cmeta_ref_parts.get('artifact_uid')
artifact_repo_alias = cmeta_ref_parts.get('artifact_repo_alias')
artifact_repo_uid = cmeta_ref_parts.get('artifact_repo_uid')
# Disambiguate repos
category_repo_artifacts = None
if (category_repo_uid is not None and category_repo_uid != '') or (category_repo_alias is not None and category_repo_alias != ''):
r = self.find_in_index('repo', self.cfg['category_repo_uid'], category_repo_alias, category_repo_uid, only_uids=True, skip_uids=skip_uids)
if r['return'] >0: return r
category_repo_artifacts = r['artifact_uids']
# Disambiguate categories (need UID and alias)
r = self.find_in_index('category', 'dd9ea50e7f76467f', category_alias, category_uid, repos = category_repo_artifacts, skip_uids=skip_uids)
if r['return'] >0: return r
category_artifacts = r['artifacts']
artifacts = []
if category_artifacts:
artifact_repo_artifacts = None
if (artifact_repo_uid is not None and artifact_repo_uid != '') or (artifact_repo_alias is not None and artifact_repo_alias != ''):
r = self.find_in_index('repo', self.cfg['category_repo_uid'], artifact_repo_alias, artifact_repo_uid, only_uids=True, skip_uids=skip_uids)
if r['return'] >0: return r
artifact_repo_artifacts = r['artifact_uids']
# Find artifacts
for category in category_artifacts:
# category_alias should always exist until we, by accident, add non-aliased category (UID)
category_alias = category['cmeta_ref_parts'].get('artifact_alias')
if category_alias is None or category_alias=="": # or category_alias=='repo':
continue
category_uid = category['cmeta_ref_parts']['artifact_uid']
category_cmeta = category['cmeta']
from_index = False
if category_cmeta.get('no_index', False) and not skip_non_indexed:
r = self.find_in_file_system(category_cmeta, category_alias, category_uid, artifact_alias, artifact_uid, repo_uids = artifact_repo_artifacts)
if r['return'] >0:
if r['return'] == 16:
# If index not found
continue
return r
else:
from_index = True
r = self.find_in_index(category_alias, category_uid, artifact_alias, artifact_uid, repos = artifact_repo_artifacts, add_index_file = add_index_file, skip_uids=skip_uids)
if r['return'] >0:
if r['return'] == 16:
# If index not found
continue
return r
# An old alias left behind by "migrate" stands for the artifact it points to
if follow_migrated and (artifact_uid is None or artifact_uid == '') and \
artifact_alias and '*' not in artifact_alias and '?' not in artifact_alias:
rr = self._follow_migrated(r['artifacts'], category_cmeta, category_alias, category_uid,
add_index_file = add_index_file)
if rr['return'] > 0: return rr
r['artifacts'] = rr['artifacts']
# Check conditions
add_artifacts = []
if all_tags is not None and len(all_tags)>0:
tmp_artifacts = []
for a in r['artifacts']:
if set(a['cmeta'].get('tags',[])) == set(all_tags):
tmp_artifacts.append(a)
else:
tmp_artifacts = r['artifacts']
if tags != None and len(tags)>0:
# Split tags into inclusion and exclusion sets
inclusion_tags = []
exclusion_tags = []
for tag in tags:
tag_str = str(tag)
if tag_str.startswith('-'):
# Remove the '-' prefix and add to exclusion list
exclusion_tags.append(tag_str[1:].lower())
else:
inclusion_tags.append(tag_str.lower())
for a in tmp_artifacts:
cmeta = a['cmeta']
ctags = cmeta.get('tags', [])
if type(ctags) != list:
cmeta_ref_parts = a['cmeta_ref_parts']
cmeta_ref_parts_artifact_alias = cmeta_ref_parts['artifact_alias']
cmeta_ref_parts_artifact_uid = cmeta_ref_parts['artifact_uid']
return {'return':1, 'error':f'tags are corrupted for artifact "{cmeta_ref_parts_artifact_alias},{cmeta_ref_parts_artifact_uid}"'}
# Convert artifact tags to lowercase for comparison
ctags_lower = [str(ctag).lower() for ctag in ctags]
# Check if all inclusion tags are present
has_all_inclusion = all(tag in ctags_lower for tag in inclusion_tags)
if not has_all_inclusion:
continue
# Check if none of the exclusion tags are present
has_no_exclusion = not any(tag in ctags_lower for tag in exclusion_tags)
if has_no_exclusion:
add_artifacts.append(a)
else:
add_artifacts = tmp_artifacts
if match is not None and len(match)>0:
add_artifacts2 = []
for a in add_artifacts:
if utils.common.matches_query(a['cmeta'],
match,
match_version_func = self.match_version_func,
match_empty_version = match_empty_version,
match_empty_values = match_empty_values,
):
add_artifacts2.append(a)
add_artifacts = add_artifacts2
# Adding artifacts: those from the index get their own copy of the meta and of its tags
# (one level: a caller that edits them - the aops selection adds a tag - edits its copy,
# not the cached index of this process; deeper values are shared and must be left as
# they are). Only the records returned pay for it, after the filters above.
if from_index:
for a in add_artifacts:
cmeta = a['cmeta']
if type(cmeta) is dict:
cmeta = dict(cmeta)
ctags = cmeta.get('tags')
if type(ctags) is list:
cmeta['tags'] = ctags[:]
a['cmeta'] = cmeta
artifacts.append(a)
else:
artifacts.extend(add_artifacts)
if len(artifacts) == 0:
x_artifact_alias = "artifacts" if artifact_alias == '' or artifact_alias == None else f'"{artifact_alias}"'
return _error(f'{category_alias} {x_artifact_alias} not found', 16, None, False) #self.fail_on_error)
return {'return':0, 'artifacts':artifacts}
###################################################################################################
def _follow_migrated(
self,
artifacts, # Artifacts found by alias in one category.
category_meta, # Metadata of that category.
category_alias, # Category alias.
category_uid, # Category UID.
add_index_file = False, # If True, include index_file path in the result.
):
"""
Replace the migration stubs among artifacts by the artifacts they point to.
A stub is what `migrate` leaves under an old alias: an artifact whose meta has
`migrated_to` (the new "alias,UID") and `migrated_when`. The target is found by its UID,
in any repository; a stub pointing to a stub is followed too, up to MAX_MIGRATED_HOPS.
Each stub followed prints one notice per process on stdout.
Args:
artifacts: Artifacts found by alias in one category.
category_meta: Metadata of that category.
category_alias: Category alias.
category_uid: Category UID.
add_index_file: If True, include index_file path in the result.
Returns:
dict: {'return': 0, 'artifacts': [...]} with the stubs replaced (and no artifact
twice), or an error if a stub points to an artifact that does not exist.
Raises:
Exception: Propagated runtime errors, if any.
"""
result = []
seen = set()
for artifact in artifacts:
current = artifact
hops = 0
while (current.get('cmeta') or {}).get('migrated_to'):
stub_meta = current['cmeta']
stub_alias = current['cmeta_ref_parts'].get('artifact_alias', current['cmeta_ref_parts'].get('artifact_uid'))
migrated_to = str(stub_meta['migrated_to'])
hops += 1
if hops > MAX_MIGRATED_HOPS:
return _error(f'{category_alias} "{stub_alias}": more than {MAX_MIGRATED_HOPS} migration stubs in a row', 1, None, self.fail_on_error)
r = utils.names.parse_cmeta_name(migrated_to)
if r['return'] > 0: return r
target_uid = r['name'].get('uid')
if not target_uid:
return _error(f'{category_alias} "{stub_alias}" was migrated to "{migrated_to}", which has no UID', 1, None, self.fail_on_error)
if category_meta.get('no_index', False):
r = self.find_in_file_system(category_meta, category_alias, category_uid, None, target_uid)
else:
r = self.find_in_index(category_alias, category_uid, None, target_uid, add_index_file = add_index_file)
if r['return'] > 0 or len(r.get('artifacts', [])) == 0:
if r['return'] > 0 and r['return'] != 16: return r
return _error(f'{category_alias} "{stub_alias}" was migrated to "{migrated_to}", which is not found', 16, None, False)
key = (category_alias, stub_alias, migrated_to)
if key not in _migrated_notices:
_migrated_notices.add(key)
when = str(stub_meta.get('migrated_when', ''))[:10]
x = f' on {when}' if when else ''
utils.common.console_line(f'{category_alias} "{stub_alias}" was migrated to "{migrated_to}"{x}; '
f'using it - please update the reference', 'warning', logger=self.logger)
current = r['artifacts'][0]
uid = current['cmeta_ref_parts'].get('artifact_uid')
if uid not in seen:
seen.add(uid)
result.append(current)
return {'return': 0, 'artifacts': result}
######################################################################################################################
[docs]
def reindex(
self,
con = False, # If True, print console messages during reindexing.
verbose = False, # If True, print detailed progress information.
):
"""
Clean index and reindex all repositories.
Rebuilds the index from scratch by scanning all repositories; the previous index
is replaced only when the new one is complete (see `index`).
Args:
con: If True, print console messages during reindexing.
verbose: If True, print detailed progress information.
Returns:
dict: Dictionary with 'return': 0 on success, or 'return' > 0 and 'error' on failure.
Raises:
Exception: Propagated runtime errors, if any.
"""
return self.index(clean=True, con=con, verbose=verbose)
[docs]
def index(
self,
clean = False, # If True, clear and rebuild index artifacts from scratch.
con = False, # If True, print output to console.
verbose = False, # If True, enable verbose output.
add_repo_paths = [], # Repository paths to add before indexing.
delete_repo_paths = [], # Repository paths to remove before indexing.
):
"""
Index repos
With `clean`, the new index is built completely in a temporary sibling folder of the index
(`index.tmp-<pid>`) and swapped in place of the old one only when it is complete, so that
an aborted reindex leaves the previous index intact and usable.
Args:
clean: If True, clear and rebuild index artifacts from scratch.
con: If True, print output to console.
verbose: If True, enable verbose output.
add_repo_paths: Repository paths to add before indexing.
delete_repo_paths: Repository paths to remove before indexing.
Returns:
dict: Operation result.
Raises:
Exception: Propagated runtime errors, if any.
"""
what = 'the full reindex' if clean else 'the index of the repositories'
note = f'{what} by pid {os.getpid()} on {utils.files._lock_host_name()} since {time.strftime("%Y-%m-%d %H:%M:%S")}'
r = self._lock_index(what, note=note)
if r['return']>0: return r
index_lock = r['lock']
try:
if not clean:
return self._index(self.index_path, clean, con, verbose, add_repo_paths, delete_repo_paths)
return self._index_full(con, verbose, add_repo_paths, delete_repo_paths)
finally:
index_lock.release()
def _index_lock_file(self):
"""The lock file of the index: `<home>/index.lock`, the sidecar of the index folder (the swap of a
full reindex leaves it in place)."""
return utils.files._get_lockfile_path(os.path.normpath(str(self.index_path)))
def _lock_index(
self,
why, # What waits, for the notice of a waiter ("the write of a record", "the full reindex" ...).
note = None, # A note left in the lock file for the waiters (who holds it and why).
):
"""
Take the index lock (`<home>/index.lock`, see INDEX_LOCK_TIMEOUT): a full reindex holds it for
its whole rebuild and swap, the index of pulled or plugged repositories for its duration and
every write of a record for a moment, so that a write never lands in an index that is about
to be replaced and two rebuilds never race for the swap. Readers do not take it. A waiter
says so on stdout after INDEX_LOCK_NOTICE_SECONDS, naming the holder when it left a note, and
gives up after INDEX_LOCK_TIMEOUT seconds (the environment variable INDEX_LOCK_TIMEOUT_ENV
changes that).
Returns:
dict: {'return': 0, 'lock': <PathLock>} - the caller releases the lock in a `finally` -
or 'return' > 0 and 'error'.
"""
lock_file = self._index_lock_file()
limit = utils.files.lock_timeout(INDEX_LOCK_TIMEOUT, INDEX_LOCK_TIMEOUT_ENV, logger=self.logger)
lock = utils.files.PathLock(lock_file, logger=self.logger)
try:
utils.files._acquire_with_notice(lock, limit, f'the index in {self.index_path}', why, INDEX_LOCK_TIMEOUT_ENV, note=note, logger=self.logger)
except TimeoutError as e:
return {'return':1, 'error': str(e)}
except Exception as e:
return {'return':1, 'error': f'cannot lock the index in {self.index_path}: {e}'}
return {'return':0, 'lock':lock}
[docs]
def lock_index(
self,
why, # What waits, for the notice of a waiter ("the creation of an artifact" ...).
note = None, # A note left in the lock file for the waiters (who holds it and why).
):
"""
Take the index lock for an operation of several steps that must not interleave with a full
reindex or with the same operation of another process (see `_lock_index`): the folder, the meta
and the record of a new artifact are made under it. Pass the lock to `add_to_index(index_lock=...)`
and release it in a `finally`.
Returns:
dict: {'return': 0, 'lock': <PathLock>} or 'return' > 0 and 'error'.
"""
return self._lock_index(why, note)
def _remove_index_leftovers(
self,
index_path, # The index folder (normalized).
):
"""Remove every `index.tmp-*` / `index.old-*` sibling of the index folder: with the index lock held
no other rebuild is running, so they are the leftovers of killed or crashed rebuilds."""
import shutil
folder = os.path.dirname(index_path) or '.'
base = os.path.basename(index_path)
try:
names = os.listdir(folder)
except OSError:
return
for name in names:
if name.startswith(base + '.tmp-') or name.startswith(base + '.old-'):
path = os.path.join(folder, name)
if os.path.isdir(path):
self.logger.debug(f'removing the leftover of an earlier rebuild of the index: {path}')
shutil.rmtree(path, ignore_errors=True)
def _index_full(
self,
con = False, # If True, print output to console.
verbose = False, # If True, enable verbose output.
add_repo_paths = [], # Repository paths to add before indexing.
delete_repo_paths = [], # Repository paths to remove before indexing.
):
"""
The full rebuild of `index` with the index lock held by the caller: the new index is built in
`index.tmp-<pid>` and swapped in place of the old one when it is complete; the leftovers of
earlier rebuilds are removed first.
"""
import shutil
index_path = os.path.normpath(str(self.index_path))
index_tmp_path = f'{index_path}.tmp-{os.getpid()}'
index_old_path = f'{index_path}.old-{os.getpid()}'
self._remove_index_leftovers(index_path)
os.makedirs(index_tmp_path)
try:
r = self._index(index_tmp_path, True, con, verbose, add_repo_paths, delete_repo_paths)
except BaseException:
shutil.rmtree(index_tmp_path, ignore_errors=True)
raise
if r['return']>0:
shutil.rmtree(index_tmp_path, ignore_errors=True)
return r
# The new index is complete: move the old one away, put the new one in its place, remove the old one
# (os.replace with the retries that Windows needs when another process still reads a file)
try:
if os.path.isdir(index_path):
utils.files._replace_file(index_path, index_old_path, self.logger)
utils.files._replace_file(index_tmp_path, index_path, self.logger)
except Exception as e:
if not os.path.isdir(index_path) and os.path.isdir(index_old_path):
os.rename(index_old_path, index_path)
shutil.rmtree(index_tmp_path, ignore_errors=True)
return {'return':1, 'error': f'failed to replace the index in {index_path} ({e}) - the previous index is kept'}
shutil.rmtree(index_old_path, ignore_errors=True)
return r
def _index(
self,
index_path, # Folder for the index files: the index itself, or an empty temporary folder for a full rebuild.
clean = False, # If True, build the index from scratch (index_path must be empty).
con = False, # If True, print output to console.
verbose = False, # If True, enable verbose output.
add_repo_paths = [], # Repository paths to add before indexing.
delete_repo_paths = [], # Repository paths to remove before indexing.
):
"""
Index repos into `index_path` (see `index`).
"""
from tqdm import tqdm
import time
time_start = time.monotonic()
conx = True if verbose else False
if conx:
print ('='*40)
if con:
print ('Reindexing all repos - it can take some time ...')
if conx:
print ('')
print (f'Index path: {index_path}')
######################################################################################################################
# Re-reading repo paths file and checking paths
repos_meta = {}
repos_config_path = self.repos_config_path
# Then checking internal repo path
this_module_path = os.path.dirname(os.path.abspath(__file__))
force_internal_repo_path = os.environ.get(self.cfg['env_var_internal_repo_path'], '').strip()
if force_internal_repo_path != '':
this_internal_repo_path = force_internal_repo_path
else:
this_internal_repo_path = os.path.join(this_module_path, 'internal-repo')
existing_internal_repo_path = None
if conx:
print (f'Repo file path: {repos_config_path}')
# Under the lock of repos.json, held until it is rewritten or left as it is: a plug or unplug of
# another process at the same moment is neither lost nor read half-written
r = utils.files.safe_read_file(repos_config_path, lock=True, keep_locked=True, retry_if_not_found=3, fail_on_error=self.fail_on_error, logger=self.logger)
if r['return']>0: return r
repos_config_lock = r['file_lock']
paths_to_repos = {} # the repositories indexed by this run
repos_to_write = {} # what repos.json lists after it: those, and - in their places - the registered
# repositories that cannot be reached now
unreachable = {} # path -> why
newly_unreachable = [] # those of them that were reachable at the run before
back = [] # paths that could not be reached at an earlier run and can now
original_paths_to_repos = r['data']
to_update = False
for path in original_paths_to_repos:
local_meta = original_paths_to_repos[path]
extra_meta = local_meta.get('meta', {})
if path.endswith('internal-repo') and os.path.normpath(path) != this_internal_repo_path:
path = this_internal_repo_path
to_update = True
path_to_repo_desc = os.path.join(path, self.cfg['repo_meta_desc'])
why_not = ''
if not os.path.isfile(path_to_repo_desc):
if extra_meta.get('keep') and 'keep_repo_meta' in local_meta:
repo_meta = local_meta['keep_repo_meta']
repos_meta[path] = repo_meta
paths_to_repos[path]={'meta': extra_meta,
'keep_repo_meta': repo_meta}
repos_to_write[path] = paths_to_repos[path]
continue
why_not = 'its folder or its ' + self.cfg['repo_meta_desc'] + ' is not there'
else:
r = utils.files.safe_read_file(path_to_repo_desc, retry_if_not_found=3, fail_on_error=self.fail_on_error, logger=self.logger)
if r['return']==0 and isinstance(r.get('data'), dict) and r['data'].get('artifact'):
repo_meta = r['data']
if len(extra_meta)>0:
repo_meta.update(extra_meta)
repos_meta[path] = repo_meta
paths_to_repos[path] = {}
if len(extra_meta)>0:
paths_to_repos[path]['meta'] = extra_meta
if 'keep_repo_meta' in local_meta:
# the stored copy of a kept repository stays with its registration (it was left out
# of the file whenever the file was rewritten while the repository was reachable)
paths_to_repos[path]['keep_repo_meta'] = local_meta['keep_repo_meta']
repos_to_write[path] = paths_to_repos[path]
if local_meta.get(self.KEY_REPO_UNREACHABLE):
back.append(path)
to_update = True
continue
if r['return']==0:
why_not = 'its ' + self.cfg['repo_meta_desc'] + ' names no artifact'
else:
why_not = 'its ' + self.cfg['repo_meta_desc'] + ' cannot be read' + (f" ({r.get('error')})" if r.get('error') else '')
# A registered repository that cannot be reached - a detached drive, a share that is down, a
# descriptor that is being edited - stays registered (it was dropped from repos.json without a
# word, and had to be plugged again) and is left out of this index
unreachable[path] = why_not
entry = dict(local_meta)
if not entry.get(self.KEY_REPO_UNREACHABLE):
entry[self.KEY_REPO_UNREACHABLE] = True
to_update = True
newly_unreachable.append(path)
repos_to_write[path] = entry
if to_update:
# Do not sort keys - preserve order!
r = utils.files.safe_write_file(repos_config_path, repos_to_write, file_lock=repos_config_lock, atomic=True, fail_on_error=self.fail_on_error, logger=self.logger, sort_keys=False)
if r['return']>0: return r
else:
r = utils.files.unlock_path(repos_config_path, repos_config_lock, fail_on_error=self.fail_on_error, logger=self.logger)
if r['return']>0: return r
for path in unreachable:
message = (f'repository "{path}" cannot be reached ({unreachable[path]}): it stays registered and is left out of the index - '
f'"cx repo reindex {path}" or "cx --reindex" brings it back when it is there again, "cx repo unplug {path}" forgets it')
utils.common.console_line(message, 'warning', logger=self.logger, con=con)
if len(back) > 0:
# Reachable again: indexed with this run, as if it had just been plugged
add_repo_paths = list(add_repo_paths) + [path for path in back if path not in add_repo_paths]
if len(newly_unreachable) > 0:
# Out of reach since the run before: its records leave the index with this run (an incremental run
# keeps the records of the repositories it does not scan; a full rebuild has none of them anyway)
delete_repo_paths = list(delete_repo_paths) + [path for path in newly_unreachable if path not in delete_repo_paths]
if len(paths_to_repos) == 0:
return {'return':1, 'error':f'could not find any repository in {repos_config_path}'}
######################################################################################################################
# Indexing repos ...
index_repo_file = os.path.join(index_path, 'repo' + self.index_extension)
index_repos = {self.KEY_INDEX_UIDS:{}, self.KEY_INDEX_LOWERCASE_ALIASES:{}}
repo_uids_to_use = []
if conx:
print ('')
print ('Indexing repositories ...')
print ('')
for path in paths_to_repos.keys():
repo_meta = repos_meta[path]
full_path = _get_full_path(path, repo_meta)
if conx:
print (f' Analyzing repository in {full_path} ...')
artifact_name = repo_meta['artifact']
r = utils.names.parse_cmeta_name(artifact_name)
if r['return']>0: return r
cmeta_name_parts = r['name']
uid = cmeta_name_parts.get('uid')
alias = cmeta_name_parts.get('alias')
lowercase_alias = alias.lower()
# # Change to proper artifact and category
# if 'artifact' not in repo_meta:
# repo_meta['artifact'] = alias + ','+uid
if 'category' not in repo_meta:
repo_meta['category'] = 'repo,' + self.cfg['category_repo_uid']
if conx:
print (f" CID = {lowercase_alias},{uid}")
repo_uids_to_use.append(uid.lower())
if lowercase_alias in index_repos[self.KEY_INDEX_LOWERCASE_ALIASES]:
print (f' Warning: repo "{alias}" is already in index!')
if uid is not None and uid in index_repos[self.KEY_INDEX_LOWERCASE_ALIASES][lowercase_alias]:
return {'return':1, 'error': f'ambiguity - repo "{alias}" with the same UID "{uid}" alredy exists in the index - please fix it!'}
index_repos[self.KEY_INDEX_LOWERCASE_ALIASES][lowercase_alias] = [uid]
entry = {'path': path, 'full_path':full_path}
cmeta_ref_parts = {'category_alias':'repo', 'category_uid':self.cfg['category_repo_uid'], 'artifact_alias':alias, 'artifact_uid':uid}
if alias != alias.lower():
cmeta_ref_parts['artifact_alias_lowercase'] = alias.lower()
entry['cmeta_ref_parts'] = cmeta_ref_parts
method = ''
path_git = os.path.join(path, '.git')
if os.path.isdir(path_git):
method = 'git'
entry['cmeta'] = repo_meta.copy()
entry['cmeta']['method'] = method
# Keep clean copy just in case
entry['_cmr'] = repo_meta
index_repos[self.KEY_INDEX_UIDS][uid] = entry
if conx:
print('')
print(f' Recording repo index file: {index_repo_file}')
# Use atomic write to avoid corrupting large index files
r = utils.files.safe_write_file(index_repo_file, index_repos, atomic=True, fail_on_error=self.fail_on_error, logger=self.logger, sort_keys=False)
if r['return']>0: return r
# A repository that asks for the soft lock is known to the path lock of this process at once (the
# processes that start later read it from this index: see init)
for path in paths_to_repos.keys():
self._register_soft_lock(path, repos_meta[path])
######################################################################################################################
# Indexing categories
index_categories = {self.KEY_INDEX_UIDS:{}, self.KEY_INDEX_LOWERCASE_ALIASES:{}}
if conx:
print ('')
print ('Indexing categories ...')
print ('')
categories = []
categories_to_index = []
for path in paths_to_repos:
repo_meta = repos_meta[path]
repo_full_path = _get_full_path(path, repo_meta)
category_full_path = os.path.join(repo_full_path, 'category')
if os.path.isdir(category_full_path):
if conx:
print (f' Processing categories in {category_full_path} ...')
category_dirs = os.listdir(category_full_path)
for category in sorted(category_dirs):
category_meta_desc_file_json = os.path.join(category_full_path, category, self.cfg['meta_filename_base'] + '.json')
category_meta_desc_file_yaml = os.path.join(category_full_path, category, self.cfg['meta_filename_base'] + '.yaml')
category_meta = {}
if os.path.isfile(category_meta_desc_file_yaml):
r = utils.files.safe_read_file(category_meta_desc_file_yaml, retry_if_not_found=3, fail_on_error=self.fail_on_error, logger=self.logger)
if r['return']==0:
category_meta = r['data']
elif os.path.isfile(category_meta_desc_file_json):
r = utils.files.safe_read_file(category_meta_desc_file_json, retry_if_not_found=3, fail_on_error=self.fail_on_error, logger=self.logger)
if r['return']==0:
category_meta = r['data']
else:
# Checking older format
category_meta_desc_file_json = os.path.join(category_full_path, category, '_cm.json')
category_meta_desc_file_yaml = os.path.join(category_full_path, category, '_cm.yaml')
if os.path.isfile(category_meta_desc_file_yaml):
r = utils.files.safe_read_file(category_meta_desc_file_yaml, retry_if_not_found=3, fail_on_error=self.fail_on_error, logger=self.logger)
if r['return']==0:
category_meta = r['data']
elif os.path.isfile(category_meta_desc_file_json):
r = utils.files.safe_read_file(category_meta_desc_file_json, retry_if_not_found=3, fail_on_error=self.fail_on_error, logger=self.logger)
if r['return']==0:
category_meta = r['data']
if category_meta:
# Update to new format
uid = category_meta['uid']
alias = category_meta.get('alias')
category_meta['artifact'] = uid
category_meta['category'] = 'category,dd9ea50e7f76467f'
for key in ['uid', 'alias', 'automation_uid', 'automation_alias']:
if key in category_meta:
del(category_meta[key])
category_meta_desc_file_yaml = os.path.join(category_full_path, category, self.cfg['meta_filename_base'] + '.yaml')
r = utils.files.safe_write_file(category_meta_desc_file_yaml, category_meta, fail_on_error=self.fail_on_error, logger=self.logger, preserve=True)
if r['return']>0: return r
if category_meta:
category_path = os.path.join(category_full_path, category)
category_name = category_meta['artifact']
# if conx:
# print (f' Found category "{category}"')
r = utils.names.parse_cmeta_name(category_name)
if r['return']>0: return r
cmeta_name_parts = r['name']
uid = cmeta_name_parts.get('uid')
alias = cmeta_name_parts.get('alias')
if alias == None:
alias = category
if alias != None:
lowercase_alias = alias.lower()
if uid is not None and uid in index_categories[self.KEY_INDEX_UIDS]:
# The same category UID in two repositories: the first copy stays in the index, this one is skipped
xpath = index_categories[self.KEY_INDEX_UIDS][uid]['path']
print (f' Warning: category "{alias}" with the same UID "{uid}" is already in the index - keeping the first copy and skipping this one:')
print (f' * kept: {xpath}')
print (f' * skipped: {category_path}')
self.logger.warning(f'duplicate category UID "{uid}": kept "{xpath}", skipped "{category_path}"')
continue
if alias is not None and lowercase_alias in index_categories[self.KEY_INDEX_LOWERCASE_ALIASES]:
print (f' Warning: category "{alias}" already exists in the index!')
category_entry = {'category':category, 'meta':category_meta}
categories.append(category_entry)
if path in add_repo_paths:
categories_to_index.append(category_entry)
# ??? FGG 20260611 - we actually need to check all existing categories and not just in the pulled repo
# because it can have artifacts with categories defined somewhere else
cmeta_ref_parts = {'category_alias':'category', 'category_uid':'dd9ea50e7f76467f', 'artifact_uid':uid}
if alias is not None and alias != '':
uids = index_categories[self.KEY_INDEX_LOWERCASE_ALIASES].get(lowercase_alias, [])
if uid not in uids:
uids.append(uid)
index_categories[self.KEY_INDEX_LOWERCASE_ALIASES][lowercase_alias] = uids
if len(uids)>1:
print (f' Warning: AMBIGUITY for category "{alias}": more than 1 UID found in paths:')
for uid in uids:
if uid in index_categories[self.KEY_INDEX_UIDS]:
print (' * ' + index_categories[self.KEY_INDEX_UIDS][uid]['path'])
print (' * ' + category_path)
if con:
# A pause for the person at the terminal; with nobody there (a detached
# job, a pipe at its end) the index goes on instead of ending in EOFError
utils.common.ask(' Fix it or press Enter to continue!', optional=True)
cmeta_ref_parts['artifact_alias'] = alias
if alias != alias.lower():
cmeta_ref_parts['artifact_alias_lowercase'] = lowercase_alias
repo_name = repo_meta['artifact']
r = utils.names.parse_cmeta_name(repo_name)
if r['return']>0: return r
repo_name_parts = r['name']
repo_alias = repo_name_parts.get('alias')
repo_uid = repo_name_parts.get('uid')
if repo_alias is not None and repo_alias != '':
cmeta_ref_parts['repo_alias'] = repo_alias
if repo_uid is not None and repo_uid != '':
cmeta_ref_parts['repo_uid'] = repo_uid
entry = {'path': category_path}
entry['cmeta_ref_parts'] = cmeta_ref_parts
entry['cmeta'] = category_meta
index_categories[self.KEY_INDEX_UIDS][uid] = entry
if index_categories[self.KEY_INDEX_UIDS]:
index_category_file = os.path.join(index_path, 'category' + self.index_extension)
if conx:
print('')
print(f' Recording category index file ({len(index_categories[self.KEY_INDEX_UIDS])} categories found): {index_category_file}')
# Use atomic write to avoid corrupting large index files
r = utils.files.safe_write_file(index_category_file, index_categories, atomic=True, fail_on_error=self.fail_on_error, logger=self.logger, sort_keys=False)
if r['return']>0: return r
# Clean removed category indexes
if conx:
print('')
print(f'Cleaning unused category files in {index_path} ...')
all_category_filenames = ['category' + self.index_extension,
'repo' + self.index_extension]
for category_mix in categories:
all_category_filenames.append(category_mix['category'] + self.index_extension)
try:
for filename in os.listdir(index_path):
if filename.endswith(self.index_extension):
if filename not in all_category_filenames:
file_path = os.path.join(index_path, filename)
if os.path.isfile(file_path):
os.remove(file_path)
except Exception as e:
if self.fail_on_error:
return {'return': 1, 'error': f'Failed to clean index files: {str(e)}'}
else:
self.logger.warning(f'Failed to clean some index files: {str(e)}')
######################################################################################################################
# Indexing artifacts
index_artifacts = {}
artifact_num = 0
if clean or len(add_repo_paths)>0:
if conx:
print ('')
print ('Indexing artifacts ...')
print ('')
# selected_paths_to_repos = add_repo_paths if len(add_repo_paths)>0 else paths_to_repos
# We go through all repos but check all or selected categories only
# selected_paths_to_repos = paths_to_repos
for path in paths_to_repos: #selected_paths_to_repos:
repo_meta = repos_meta[path]
repo_name = repo_meta['artifact']
r = utils.names.parse_cmeta_name(repo_name)
if r['return']>0: return r
cmeta_name_parts = r['name']
repo_uid = cmeta_name_parts.get('uid')
repo_alias = cmeta_name_parts.get('alias')
repo_full_path = _get_full_path(path, repo_meta)
if conx:
print (f' Processing repo in {repo_full_path} ...')
selected_categories = categories_to_index if len(add_repo_paths)>0 and path not in add_repo_paths else categories
# selected_categories = categories_to_index if len(add_repo_paths)>0 else categories
for category_mix in selected_categories:
category = category_mix['category']
# Skip already index categories (repo and category)
if category in ['category', 'repo']:
continue
category_meta = category_mix['meta']
if category_meta.get('no_index', False):
continue
category_name = category_meta['artifact']
r = utils.names.parse_cmeta_name(category_name)
if r['return']>0: return r
cmeta_name_parts = r['name']
category_uid = cmeta_name_parts.get('uid')
category_alias = category
full_category_name = f'{category},' + category_uid
path_to_category = os.path.join(repo_full_path, category)
r = self._find_artifacts(repo_meta, repo_alias, repo_uid, category_meta, category_alias, category_uid, path_to_category,
con, conx, index_artifacts, artifact_num)
if r['return'] >0: return r
an = r['artifact_num']
artifact_num = an
if not clean:
# The records of the repositories being indexed again are removed from every existing
# category index below, also from the index of a category of which such a repository
# holds no artifact any more (otherwise the records of its deleted artifacts would stay)
for category_mix in categories:
category = category_mix['category']
if category in ['category', 'repo'] or category_mix['meta'].get('no_index', False):
continue
if category not in index_artifacts and os.path.isfile(os.path.join(index_path, category + self.index_extension)):
index_artifacts[category] = {self.KEY_INDEX_UIDS:{}, self.KEY_INDEX_LOWERCASE_ALIASES:{}}
else:
for category_mix in categories:
index_artifacts[category_mix['category']] = {}
if index_artifacts:
if conx:
print ('')
print (f'Recording index files for artifacts in {index_path} ...')
for category in tqdm(sorted(index_artifacts), disable = not (con and conx), desc=" Recording index file: "):
category_index = index_artifacts[category]
index_artifact_file = os.path.join(index_path, category + self.index_extension)
# if conx:
# print(f' Recording {category} index file: {index_artifact_file}')
existing_category_index = {}
index_file_lock = None
if clean:
existing_category_index = category_index
else:
if os.path.isfile(index_artifact_file):
r = utils.files.safe_read_file(index_artifact_file, lock=True, keep_locked=True, fail_on_error=self.fail_on_error, logger=self.logger)
if r['return']>0: return r
existing_category_index = r['data']
index_file_lock = r['file_lock']
if len(add_repo_paths)>0 or len(delete_repo_paths)>0:
uids = existing_category_index.setdefault(self.KEY_INDEX_UIDS, {})
lowercase_aliases = existing_category_index.setdefault(self.KEY_INDEX_LOWERCASE_ALIASES, {})
# Remove UIDs of old/updated repos
for uid in list(uids.keys()):
artifact = uids[uid]
path = artifact['path']
for remove_path in add_repo_paths + delete_repo_paths:
if utils.files.is_path_within(remove_path, path):
del(uids[uid])
break
# Remove aliases
updated_uids_keys = list(uids.keys())
for lowercase_alias in list(lowercase_aliases.keys()):
for uid in lowercase_aliases[lowercase_alias]:
if uid not in updated_uids_keys:
del(lowercase_aliases[lowercase_alias])
break
if len(add_repo_paths)>0:
# Merge new ones
new_uids = category_index.get(self.KEY_INDEX_UIDS, {})
new_lowercase_aliases = category_index.get(self.KEY_INDEX_LOWERCASE_ALIASES, {})
for uid in new_uids:
uids[uid] = new_uids[uid]
for lowercase_alias in new_lowercase_aliases:
if lowercase_alias not in lowercase_aliases:
lowercase_aliases[lowercase_alias] = []
for xuid in new_lowercase_aliases[lowercase_alias]:
if xuid not in lowercase_aliases[lowercase_alias]:
lowercase_aliases[lowercase_alias].append(xuid)
artifact_num += len(existing_category_index.get(self.KEY_INDEX_UIDS, {}))
# Atomic for a new index file too: a reader never sees it half-written
r = utils.files.safe_write_file(index_artifact_file, existing_category_index, file_lock=index_file_lock,
atomic=True, fail_on_error=self.fail_on_error, logger=self.logger, sort_keys=False)
if r['return']>0: return r
time_end = time.monotonic()
elapsed = time_end - time_start
if conx:
print ('')
print (f'Number of index artifacts: {artifact_num}')
print (f'Indexing time: {elapsed:.2f} sec.')
print ('='*40)
r = {'return':0, 'elapsed_time':elapsed}
return r
################################################################################
def _find_artifacts(
self,
repo_meta, # Repository metadata dictionary.
repo_alias, # Repository alias.
repo_uid, # Repository UID.
category_meta, # Category metadata dictionary.
category_alias, # Category alias.
category_uid, # Category UID.
path_to_category, # Path to category directory.
con, # If True, enable console output.
conx, # If True, enable extended console output.
index_artifacts, # Dictionary to populate with found artifacts (None for searching without indexing).
artifact_num, # Current artifact count.
artifact_alias = None, # Optional artifact alias to filter results (supports wildcards).
artifact_uid = None, # Optional artifact UID to filter results.
):
"""
Find and index artifacts in a category directory.
Args:
repo_meta: Repository metadata dictionary.
repo_alias: Repository alias.
repo_uid: Repository UID.
category_meta: Category metadata dictionary.
category_alias: Category alias.
category_uid: Category UID.
path_to_category: Path to category directory.
con: If True, enable console output.
conx: If True, enable extended console output.
index_artifacts: Dictionary to populate with found artifacts (None for searching without indexing).
artifact_num: Current artifact count.
artifact_alias: Optional artifact alias to filter results (supports wildcards).
artifact_uid: Optional artifact UID to filter results.
Returns:
dict: Dictionary with 'return': 0, 'artifacts' list, and 'artifact_num' on success,
or 'return' > 0 and 'error' on failure.
Raises:
Exception: Propagated runtime errors, if any.
"""
from tqdm import tqdm
artifacts = []
category = category_alias
if category_uid in repo_meta.get('sharding_slices', {}):
sharding_slices = repo_meta['sharding_slices'][category_uid]
else:
sharding_slices = category_meta.get('sharding_slices')
if os.path.isdir(path_to_category):
if conx:
print (f' Processing category {category} ...', flush=True)
if index_artifacts is not None:
# TBD sharding
if sharding_slices is not None:
# Pass artifact_alias for smart shard pruning when not indexing
search_alias = artifact_alias if index_artifacts is None else None
artifact_dirs = _get_artifacts_from_sharded_path(path_to_category, sharding_slices, artifact_alias=search_alias)
else:
artifact_dirs = os.listdir(path_to_category)
tmp_artifact_dirs = tqdm(artifact_dirs, disable = not (con and conx), desc=" Indexing artifacts: ")
else:
# Filter artifact_dirs if searching without index
tmp_artifact_dirs = []
if artifact_alias is None or artifact_alias == '':
artifact_alias = '*'
artifact_alias_lowercase = artifact_alias.lower()
if '*' in artifact_alias or '?' in artifact_alias:
if sharding_slices is not None:
tmp_artifact_dirs = _get_artifacts_from_sharded_path(path_to_category, sharding_slices, artifact_alias=artifact_alias)
else:
# Only the folders the pattern names (case-insensitively, as the sharded search does)
tmp_artifact_dirs = [d for d in os.listdir(path_to_category) if fnmatch.fnmatch(d.lower(), artifact_alias_lowercase)]
else:
last_artifact_dirs_parent = None
path_to_category_with_shard = path_to_category
if sharding_slices is not None:
r = utils.files.apply_sharding_to_path(None, artifact_alias, sharding_slices)
if r['return']>0: return r
last_artifact_dirs = r['sharded_parts']
if len(last_artifact_dirs)>0:
last_artifact_dirs_parent = os.path.join(*(last_artifact_dirs[:-1]))
path_to_category_with_shard = os.path.join(path_to_category, last_artifact_dirs_parent)
if os.path.isdir(path_to_category_with_shard):
# Every folder that spells the alias (two spellings can coexist on a case-sensitive file system)
for artifact_dir in os.listdir(path_to_category_with_shard):
if artifact_dir.lower() == artifact_alias_lowercase:
tmp_dir = artifact_dir if last_artifact_dirs_parent is None else os.path.join(last_artifact_dirs_parent, artifact_dir)
tmp_artifact_dirs.append(tmp_dir)
for long_artifact in tmp_artifact_dirs:
artifact = os.path.basename(long_artifact)
path_to_artifact = os.path.join(path_to_category, long_artifact)
tt1 = time.perf_counter()
artifact_meta_desc_file_json = os.path.join(path_to_artifact, self.cfg['meta_filename_base'] + '.json')
artifact_meta_desc_file_yaml = os.path.join(path_to_artifact, self.cfg['meta_filename_base'] + '.yaml')
artifact_meta = {}
if os.path.isfile(artifact_meta_desc_file_yaml):
# import yaml
# with open(artifact_meta_desc_file_yaml, "r") as f:
# artifact_meta = yaml.safe_load(f)
r = utils.files.safe_read_file(artifact_meta_desc_file_yaml, retry_if_not_found=3, fail_on_error=self.fail_on_error, logger=self.logger)
if r['return']==0:
artifact_meta = r['data']
elif os.path.isfile(artifact_meta_desc_file_json):
# import json
# with open(artifact_meta_desc_file_json, "r") as f:
# artifact_meta = json.load(f)
r = utils.files.safe_read_file(artifact_meta_desc_file_json, retry_if_not_found=3, fail_on_error=self.fail_on_error, logger=self.logger)
if r['return']==0:
artifact_meta = r['data']
tt2 = time.perf_counter()
if artifact_meta:
if index_artifacts is None:
# A meta without a category (or with one that does not parse) is not an artifact of this category
if 'category' not in artifact_meta:
print ('', flush=True)
print (f" Warning: {artifact} doesn't have 'category' key in {path_to_artifact}")
self.logger.error (f" Warning: {artifact} doesn't have proper 'category' key in {path_to_artifact}")
continue
r = utils.names.parse_cmeta_name(artifact_meta['category'])
if r['return']>0: return r
cmeta_category_name_parts = r['name']
if cmeta_category_name_parts['uid'] != category_uid:
continue
cmeta_category_alias = cmeta_category_name_parts.get('alias')
if cmeta_category_alias is not None and cmeta_category_alias !='' and cmeta_category_alias != category_alias:
continue
if 'artifact' not in artifact_meta:
print ('', flush=True)
print (f" Warning: {artifact} doesn't have 'artifact' key in {path_to_artifact}")
self.logger.error (f" Warning: {artifact} doesn't have proper 'artifact' key in {path_to_artifact}")
# TBD - better handling?
continue
artifact_name = artifact_meta['artifact']
r = utils.names.parse_cmeta_name(artifact_name)
if r['return']>0: return r
cmeta_name_parts = r['name']
uid = cmeta_name_parts.get('uid')
if uid is None or not utils.names.is_valid_cmeta_uid(uid):
print ('', flush=True)
print (f" Warning: {artifact} doesn't have proper {uid}")
self.logger.error (f" Warning: {artifact} doesn't have proper {uid}")
# TBD - better handling?
continue
if artifact_uid is not None and artifact_uid != uid:
continue
alias = artifact
lowercase_alias = alias.lower()
entry = {'path':path_to_artifact, 'cmeta':artifact_meta}
cmeta_ref_parts = {'artifact_uid':uid, 'category_uid':category_uid, 'repo_uid':repo_uid}
if alias is not None and alias != "":
cmeta_ref_parts['artifact_alias'] = alias
if alias != alias.lower():
cmeta_ref_parts['artifact_alias_lowercase'] = alias.lower()
if category_alias is not None and category_alias!="":
cmeta_ref_parts['category_alias'] = category_alias
if repo_alias is not None and repo_alias!="":
cmeta_ref_parts['repo_alias'] = repo_alias
entry['cmeta_ref_parts'] = cmeta_ref_parts
if sharding_slices:
entry['sharding_slices_num'] = len(sharding_slices)
if index_artifacts is None:
artifacts.append(entry)
else:
artifact_num += 1
if category not in index_artifacts:
index_artifacts[category] = {self.KEY_INDEX_UIDS:{}, self.KEY_INDEX_LOWERCASE_ALIASES:{}}
uids = index_artifacts[category][self.KEY_INDEX_UIDS]
aliases_lower_case = index_artifacts[category][self.KEY_INDEX_LOWERCASE_ALIASES]
if uid in uids:
xpath = uids[uid]['path']
if xpath != path_to_artifact:
return {'return':1, 'error': f'ambiguity - artifact "{artifact}" with the same UID "{uid}" and path "{path_to_artifact}" alredy exists in the index in path "{xpath}"- please fix it!'}
else:
uids[uid] = entry
if alias is not None and alias != '':
alias_lower_case = alias.lower()
if alias_lower_case not in aliases_lower_case:
aliases_lower_case[alias_lower_case] = []
name_uids = aliases_lower_case[alias_lower_case]
if uid not in name_uids:
name_uids.append(uid)
if len(name_uids)>1:
print ('', flush=True)
print (f' Warning: Conflict for {category_alias}:{alias} - multiple UIDs: "{name_uids} ..."')
tt3 = time.perf_counter()
# print (f' time1 = {tt2-tt1:.4f} ; time2 = {tt3-tt2:.4f}')
return {'return':0, 'artifact_num': artifact_num, 'artifacts': artifacts}
################################################################################
def _get_full_path(
path, # Base repository path
repo_meta, # Repository metadata dictionary
):
"""
Get path with prefix if specified in repo metadata
Args:
path: Base repository path
repo_meta: Repository metadata dictionary
Returns:
str: Path with prefix applied if exists, otherwise original path
Raises:
Exception: Propagated runtime errors, if any.
"""
full_path = path
subdir = repo_meta.get('subdir')
if subdir is not None:
subdir = subdir.strip()
if subdir != '':
full_path = os.path.join(path, subdir)
return full_path
################################################################################
def _get_artifacts_from_sharded_path(
base_path, # Base path to search
slices, # List of slice lengths (e.g. [2] or [3,2])
artifact_alias = None, # Optional alias to filter (supports wildcards)
):
"""
Get list of artifact directories from sharded path structure.
Args:
base_path: Base path to search
slices: List of slice lengths (e.g. [2] or [3,2])
artifact_alias: Optional alias to filter (supports wildcards)
Returns:
list: List of relative paths to artifact directories
Raises:
Exception: Propagated runtime errors, if any.
"""
if not os.path.isdir(base_path):
return []
# Normalize artifact_alias for case-insensitive matching
artifact_alias_lowercase = None
use_wildcard = False
if artifact_alias is not None:
artifact_alias_lowercase = artifact_alias.lower()
use_wildcard = '*' in artifact_alias or '?' in artifact_alias
# Calculate total depth from slices
total_depth = sum(slices)
def _recursive_traverse(
current_path, # Current directory path
current_depth, # Current depth in the shard structure
rel_path_parts, # List of path components for building relative path
):
"""
Recursively traverse sharded structure and collect matching artifacts.
Args:
current_path: Current directory path
current_depth: Current depth in the shard structure
rel_path_parts: List of path components for building relative path
Returns:
list: Matched artifact paths
Raises:
Exception: Propagated runtime errors, if any.
"""
matches = []
try:
entries = os.listdir(current_path)
except (OSError, PermissionError):
return matches
# At target depth - check for matching artifact directories
if current_depth == total_depth:
for dirname in entries:
dir_path = os.path.join(current_path, dirname)
if not os.path.isdir(dir_path):
continue
final_name = dirname.lower()
# Filter by artifact_alias if provided
if artifact_alias_lowercase is not None:
if use_wildcard:
if not fnmatch.fnmatch(final_name, artifact_alias_lowercase):
continue
else:
if final_name != artifact_alias_lowercase:
continue
# Build relative path
artifact_path = os.path.join(*(rel_path_parts + [dirname]))
matches.append(artifact_path)
# Not at target depth yet - continue traversing
elif current_depth < total_depth:
for dirname in entries:
dir_path = os.path.join(current_path, dirname)
if os.path.isdir(dir_path):
# Recursively traverse subdirectory
matches.extend(_recursive_traverse(
dir_path,
current_depth + len(dirname),
rel_path_parts + [dirname]
))
return matches
# Start recursive traversal from base_path
return _recursive_traverse(base_path, 0, [])