Sync

Core synchronization logic. sync_packages() is the public entry point — it orchestrates Gitea queries, version checks, downloads, and Devpi uploads for all configured mappings.

devpi_gitea_sync.sync

SyncFailure

Bases: RuntimeError

Raised when any part of the synchronization fails.

Source code in devpi_gitea_sync/sync.py

class SyncFailure(RuntimeError):
    """Raised when any part of the synchronization fails."""

sync_packages(config, *, organizations=None, mapping_names=None, dry_run=False, force=False, package_filters=None)

Synchronise Python packages from configured Gitea organisations into Devpi.

Parameters:

Name	Type	Description	Default
config	SyncConfig	The synchronization configuration.	required
organizations	Sequence[str] | None	A list of organizations to synchronize.	None
mapping_names	Sequence[str] | None	A list of mapping names to synchronize.	None
dry_run	bool	If True, discover assets without downloading or uploading them.	False
force	bool	If True, re-upload package versions even if they already exist in Devpi.	False
package_filters	dict[str, set[str]] | None	A dictionary of package filters.	None

Raises:

Type	Description
SyncFailure	If any part of the synchronization fails.

Source code in devpi_gitea_sync/sync.py

def sync_packages(
    config: SyncConfig,
    *,
    organizations: Sequence[str] | None = None,
    mapping_names: Sequence[str] | None = None,
    dry_run: bool = False,
    force: bool = False,
    package_filters: dict[str, set[str]] | None = None,
) -> None:
    """Synchronise Python packages from configured Gitea organisations into Devpi.

    :param config: The synchronization configuration.
    :param organizations: A list of organizations to synchronize.
    :param mapping_names: A list of mapping names to synchronize.
    :param dry_run: If True, discover assets without downloading or uploading them.
    :param force: If True, re-upload package versions even if they already exist in Devpi.
    :param package_filters: A dictionary of package filters.
    :raises SyncFailure: If any part of the synchronization fails.
    """
    selected_orgs: set[str] | None = set(organizations) if organizations else None
    selected_mappings: set[str] | None = set(mapping_names) if mapping_names else None
    normalized_filters: dict[str, set[str]] | None = None
    if package_filters:
        normalized_filters = {
            org: {canonicalize_name(pkg) for pkg in pkgs} for org, pkgs in package_filters.items()
        }
    filter_orgs: set[str] | None = set(normalized_filters.keys()) if normalized_filters else None

    failures: list[str] = []
    gitea_clients: dict[str, GiteaClient] = {}
    devpi_sessions: dict[str, DevpiSession] = {}
    download_dir = config.download_dir
    if not dry_run:
        download_dir.mkdir(parents=True, exist_ok=True)

    try:
        for mapping in config.mappings:
            if selected_orgs and mapping.organization not in selected_orgs:
                logger.debug("Skipping organization '%s' (not selected)", mapping.organization)
                continue
            if selected_mappings and mapping.name not in selected_mappings:
                logger.debug("Skipping mapping '%s' (not selected)", mapping.name)
                continue
            if filter_orgs and mapping.organization not in filter_orgs:
                logger.debug(
                    "Skipping organization '%s' (no package filters match)", mapping.organization
                )
                continue

            gitea_token = mapping.gitea_token
            if not gitea_token:
                logger.error(
                    "No Gitea token available for organization '%s'; skipping",
                    mapping.organization,
                )
                failures.append(f"{mapping.organization}: missing Gitea token")
                continue

            gitea_client = gitea_clients.get(gitea_token)
            if gitea_client is None:
                logger.debug("Creating new Gitea client for organization '%s'", mapping.organization)
                gitea_client = GiteaClient(config.gitea, gitea_token)
                gitea_clients[gitea_token] = gitea_client

            devpi_server = config.devpi_servers[mapping.devpi_server]
            devpi_session = devpi_sessions.get(mapping.devpi_server)
            if devpi_session is None:
                logger.debug("Creating new Devpi session for server '%s'", mapping.devpi_server)
                devpi_session = DevpiSession(devpi_server)
                devpi_sessions[mapping.devpi_server] = devpi_session

            try:
                _sync_org(
                    mapping=mapping,
                    gitea=gitea_client,
                    devpi=devpi_session,
                    dry_run=dry_run,
                    download_dir=download_dir,
                    force=force,
                    package_filters=normalized_filters.get(mapping.organization)
                    if normalized_filters
                    else None,
                )
            except (GiteaError, DevpiError) as exc:
                msg = f"{mapping.organization}: {exc}"
                logger.error(
                    "Synchronization for mapping '%s' failed: %s",
                    mapping.name,
                    exc,
                )
                failures.append(msg)
    finally:
        for client in gitea_clients.values():
            client.close()
        for session in devpi_sessions.values():
            session.close()

    if failures:
        error_text = "; ".join(failures)
        raise SyncFailure(f"One or more organizations failed to sync: {error_text}")