data/model/oci/tag.py
def retarget_tag(
    tag_name, manifest_id, is_reversion=False, now_ms=None, raise_on_error=False,
):
    """
    Creates or updates a tag with the specified name to point to the given manifest under its
    repository.

    If this action is a reversion to a previous manifest, is_reversion should be set to True.
    Returns the newly created tag row or None on error.
    """
    try:
        manifest = (
            Manifest.select(Manifest, MediaType)
            .join(MediaType)
            .where(Manifest.id == manifest_id)
            .get()
        )
    except Manifest.DoesNotExist:
        if raise_on_error:
            raise RetargetTagException("Manifest requested no longer exists")

        return None

    # CHECK: Make sure that we are not mistargeting a schema 1 manifest to a tag with a different
    # name.
    if manifest.media_type.name in DOCKER_SCHEMA1_CONTENT_TYPES:
        try:
            parsed = DockerSchema1Manifest(
                Bytes.for_string_or_unicode(manifest.manifest_bytes), validate=False
            )
            if parsed.tag != tag_name:
                logger.error(
                    "Tried to re-target schema1 manifest with tag `%s` to tag `%s",
                    parsed.tag,
                    tag_name,
                )
                return None
        except MalformedSchema1Manifest as msme:
            logger.exception("Could not parse schema1 manifest")
            if raise_on_error:
                raise RetargetTagException(msme)

            return None

    legacy_image = get_legacy_image_for_manifest(manifest)
    now_ms = now_ms or get_epoch_timestamp_ms()
    now_ts = int(now_ms / 1000)

    with db_transaction():
        # Lookup an existing tag in the repository with the same name and, if present, mark it
        # as expired.
        existing_tag = get_tag(manifest.repository_id, tag_name)
        if existing_tag is not None:
            _, okay = set_tag_end_ms(existing_tag, now_ms)

            # TODO: should we retry here and/or use a for-update?
            if not okay:
                return None

        # Create a new tag pointing to the manifest with a lifetime start of now.
        created = Tag.create(
            name=tag_name,
            repository=manifest.repository_id,
            lifetime_start_ms=now_ms,
            reversion=is_reversion,
            manifest=manifest,
            tag_kind=Tag.tag_kind.get_id("tag"),
        )

        return created

    Creates or updates a tag with the specified name to point to the given manifest under its
    repository.

    If this action is a reversion to a previous manifest, is_reversion should be set to True.
    Returns the newly created tag row or None on error.
    
Creates or updates a tag with the specified name to point to the given manifest under its
repository.

If this action is a reversion to a previous manifest, is_reversion should be set to True.
Returns the newly created tag row or None on error.
Creates or updates a tag with the specified name to point to the given manifest under its repository . If this action is a reversion to a previous manifest is_reversion should be set to True . Returns the newly created tag row or None on error .
def retarget_tag(
    tag_name, manifest_id, is_reversion=False, now_ms=None, raise_on_error=False,
):
    try:
        manifest = (
            Manifest.select(Manifest, MediaType)
            .join(MediaType)
            .where(Manifest.id == manifest_id)
            .get()
        )
    except Manifest.DoesNotExist:
        if raise_on_error:
            raise RetargetTagException("Manifest requested no longer exists")
        return None
    if manifest.media_type.name in DOCKER_SCHEMA1_CONTENT_TYPES:
        try:
            parsed = DockerSchema1Manifest(
                Bytes.for_string_or_unicode(manifest.manifest_bytes), validate=False
            )
            if parsed.tag != tag_name:
                logger.error(
                    "Tried to re-target schema1 manifest with tag `%s` to tag `%s",
                    parsed.tag,
                    tag_name,
                )
                return None
        except MalformedSchema1Manifest as msme:
            logger.exception("Could not parse schema1 manifest")
            if raise_on_error:
                raise RetargetTagException(msme)
            return None
    legacy_image = get_legacy_image_for_manifest(manifest)
    now_ms = now_ms or get_epoch_timestamp_ms()
    now_ts = int(now_ms / 1000)
    with db_transaction():
        existing_tag = get_tag(manifest.repository_id, tag_name)
        if existing_tag is not None:
            _, okay = set_tag_end_ms(existing_tag, now_ms)
            if not okay:
                return None
        created = Tag.create(
            name=tag_name,
            repository=manifest.repository_id,
            lifetime_start_ms=now_ms,
            reversion=is_reversion,
            manifest=manifest,
            tag_kind=Tag.tag_kind.get_id("tag"),
        )
        return created
def retarget_tag ( tag_name , manifest_id , is_reversion = False , now_ms = None , raise_on_error = False , ) : try : manifest = ( Manifest . select ( Manifest , MediaType ) . join ( MediaType ) . where ( Manifest . id == manifest_id ) . get ( ) ) except Manifest . DoesNotExist : if raise_on_error : raise RetargetTagException ( "Manifest requested no longer exists" ) return None if manifest . media_type . name in DOCKER_SCHEMA1_CONTENT_TYPES : try : parsed = DockerSchema1Manifest ( Bytes . for_string_or_unicode ( manifest . manifest_bytes ) , validate = False ) if parsed . tag != tag_name : logger . error ( "Tried to re-target schema1 manifest with tag `%s` to tag `%s" , parsed . tag , tag_name , ) return None except MalformedSchema1Manifest as msme : logger . exception ( "Could not parse schema1 manifest" ) if raise_on_error : raise RetargetTagException ( msme ) return None legacy_image = get_legacy_image_for_manifest ( manifest ) now_ms = now_ms or get_epoch_timestamp_ms ( ) now_ts = int ( now_ms / 1000 ) with db_transaction ( ) : existing_tag = get_tag ( manifest . repository_id , tag_name ) if existing_tag is not None : _ , okay = set_tag_end_ms ( existing_tag , now_ms ) if not okay : return None created = Tag . create ( name = tag_name , repository = manifest . repository_id , lifetime_start_ms = now_ms , reversion = is_reversion , manifest = manifest , tag_kind = Tag . tag_kind . get_id ( "tag" ) , ) return created
Creates or updates a tag with the specified name to point to the given manifest under its
repository.
Creates or updates a tag with the specified name to point to the given manifest under its repository .
"""
    Creates or updates a tag with the specified name to point to the given manifest under its
    repository.

    If this action is a reversion to a previous manifest, is_reversion should be set to True.
    Returns the newly created tag row or None on error.
    """ # CHECK: Make sure that we are not mistargeting a schema 1 manifest to a tag with a different # name. # Lookup an existing tag in the repository with the same name and, if present, mark it # as expired. # TODO: should we retry here and/or use a for-update? # Create a new tag pointing to the manifest with a lifetime start of now.
tag_name
manifest_id
is_reversion
now_ms
raise_on_error