cmeta.repos module

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.

Classes

class cmeta.repos.Repos(cfg: Dict[str, Any], home_path: Path, index_path: Path, repos_config_path: Path, logger: Logger = None, index_extension: str = '.pkl', fail_on_error: bool = False, match_version_func=None)[source]

Bases: object

cMeta repositories manager.

__init__(cfg: Dict[str, Any], home_path: Path, index_path: Path, repos_config_path: Path, logger: Logger = None, index_extension: str = '.pkl', fail_on_error: bool = False, match_version_func=None)[source]

Initialize Repos manager.

Parameters:

  • 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:

Operation result.

Return type:

dict

Raises:

Exception – Propagated runtime errors, if any.

init(con: bool = False, verbose: bool = False)[source]

Initialize repositories and index if running for the first time.

Creates repos directory, local repository, and triggers reindexing if needed.

Parameters:

  • con (bool) – Console mode flag (for compatibility).

  • verbose (bool) – Enable verbose output (for compatibility).

Returns:

Dictionary with ‘return’: 0 on success, or ‘return’ > 0 and ‘error’ on failure.

Return type:

dict

Raises:

Exception – Propagated runtime errors, if any.

add_to_index(cmeta, cmeta_ref_parts, path, original_alias=None, original_uid=None, sharding_slices_num=0, clean_aliases=False, index_lock=None)[source]

Insert or update an artifact record in the category index (under the index lock, see _lock_index).

Parameters:

  • 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:

Operation result, with ‘dropped_aliases’: the (lowercase) aliases dropped by clean_aliases.

Return type:

dict

Raises:

Exception – Propagated runtime errors, if any.

remove_from_index(index_file, artifact_uid, artifact_alias_lowercase)[source]

Remove an artifact from the repository index.

Parameters:

  • 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:

Dictionary with ‘return’: 0 on success, or ‘return’ > 0 and ‘error’ on failure.

Return type:

dict

Raises:

Exception – Propagated runtime errors, if any.

find_in_index(category_alias, category_uid, artifact_alias=None, artifact_uid=None, repos=[], only_uids=False, add_index_file=False, skip_uids=False)[source]

Find artifacts in the repository index.

Parameters:

  • 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:

Dictionary with ‘return’: 0 and ‘lst’ containing found artifacts,

or ‘return’ > 0 and ‘error’ on failure.

Return type:

dict

Raises:

Exception – Propagated runtime errors, if any.

find_in_file_system(category_meta, category_alias, category_uid, artifact_alias=None, artifact_uid=None, repo_uids=[])[source]

Find artifacts by scanning directories instead of using index. Used for categories with no_index flag.

Parameters:

  • 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:

{‘return’: 0, ‘artifacts’: […]} or error

Return type:

dict

Raises:

Exception – Propagated runtime errors, if any.

find(cmeta_ref: str, add_index_file: bool = False, tags: str = None, skip_uids: bool = False, skip_non_indexed: bool = False, match: bool = None, match_empty_version: bool = False, match_empty_values: bool = False, all_tags: str = None, follow_migrated: bool = True)[source]

Find artifacts by cMeta reference.

Parameters:

  • 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:

Dictionary with ‘return’: 0 and ‘artifacts’ list on success,

or ‘return’ > 0 and ‘error’ on failure.

Return type:

dict

Raises:

Exception – Propagated runtime errors, if any.

reindex(con=False, verbose=False)[source]

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).

Parameters:

  • con – If True, print console messages during reindexing.

  • verbose – If True, print detailed progress information.

Returns:

Dictionary with ‘return’: 0 on success, or ‘return’ > 0 and ‘error’ on failure.

Return type:

dict

Raises:

Exception – Propagated runtime errors, if any.

index(clean=False, con=False, verbose=False, add_repo_paths=[], delete_repo_paths=[])[source]

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.

Parameters:

  • 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:

Operation result.

Return type:

dict

Raises:

Exception – Propagated runtime errors, if any.

lock_index(why, note=None)[source]

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:

{‘return’: 0, ‘lock’: <PathLock>} or ‘return’ > 0 and ‘error’.

Return type:

dict

Constants

cmeta.repos.INDEX_LOCK_NOTICE_SECONDS = 3

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.__int__(). For floating-point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by ‘+’ or ‘-’ and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer iteral. >>> int(‘0b100’, base=0) 4

cmeta.repos.INDEX_LOCK_TIMEOUT = 600

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.__int__(). For floating-point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by ‘+’ or ‘-’ and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer iteral. >>> int(‘0b100’, base=0) 4

cmeta.repos.INDEX_LOCK_TIMEOUT_ENV = 'CMETA_INDEX_LOCK_TIMEOUT'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to ‘utf-8’. errors defaults to ‘strict’.

cmeta.repos.MAX_MIGRATED_HOPS = 8

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.__int__(). For floating-point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by ‘+’ or ‘-’ and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer iteral. >>> int(‘0b100’, base=0) 4