gitea.actions.async_run_management

async_run_management

Acting on an Actions workflow run: cancelling, approving, rerunning, deleting.

The rest of the Actions resource reads. These four write, and what they answer with is worth knowing before calling them:

• Cancelling and approving answer 200 with the run as it now stands, so the result says what the request did rather than only that it was accepted.
• Rerunning answers 201 with the run, except for the failed-jobs form, which answers 201 with no body at all. An empty result there is a rerun that started, not one that failed.
• Deleting answers 204, as a delete does everywhere in this API.

None of them has an owner-wide form: a run belongs to a repository, and so does every one of these.

The asynchronous mirror of gitea.actions.run_management. The endpoints, the arguments and the answers are that module's, and it is the one to read for what each method does and why. The difference here is aiohttp in place of requests, and the awaits that come with it.

Classes

gitea.actions.async_run_management.AsyncRunManagement

AsyncRunManagement(client: AsyncClientProtocol)

Bases: BaseActions, AsyncResource

The Actions endpoints that act on a workflow run.

Source code in src/gitea/resource/async_resource.py

def __init__(self, client: AsyncClientProtocol) -> None:
    """Initialize the Resource with a AsyncGitea client.

    Args:
        client: An instance of the AsyncGitea client.

    """
    self.client = client

Methods:

gitea.actions.async_run_management.AsyncRunManagement.cancel_workflow_run async

cancel_workflow_run(
    owner: str,
    repository: str,
    run_id: int,
    force: bool = False,
    **kwargs: Any,
) -> tuple[dict[str, Any], dict[str, Any]]

Cancel a workflow run.

Cancelling asks the run's jobs to stop and waits for them to notice, which a job whose runner has gone away never does. force marks the run cancelled regardless - a different endpoint, not a retry of the same one - and is what gets a run stuck in in_progress out of the way.

Parameters:

Name	Type	Description	Default
owner	str	The owner of the repository.	required
repository	str	The name of the repository.	required
run_id	int	The ID of the run.	required
force	bool	Whether to mark the run cancelled without waiting for its jobs to stop.	False
**kwargs	Any	Additional arguments for the request.	{}

Returns:

Type	Description
dict[str, Any]	A tuple containing the run as it now stands - its status is what
dict[str, Any]	says whether the cancellation has taken effect yet - and a dictionary
tuple[dict[str, Any], dict[str, Any]]	with metadata.

Source code in src/gitea/actions/async_run_management.py

async def cancel_workflow_run(
    self, owner: str, repository: str, run_id: int, force: bool = False, **kwargs: Any
) -> tuple[dict[str, Any], dict[str, Any]]:
    """Cancel a workflow run.

    Cancelling asks the run's jobs to stop and waits for them to notice,
    which a job whose runner has gone away never does. `force` marks the run
    cancelled regardless - a different endpoint, not a retry of the same one -
    and is what gets a run stuck in `in_progress` out of the way.

    Args:
        owner: The owner of the repository.
        repository: The name of the repository.
        run_id: The ID of the run.
        force: Whether to mark the run cancelled without waiting for its
            jobs to stop.
        **kwargs: Additional arguments for the request.

    Returns:
        A tuple containing the run as it now stands - its `status` is what
        says whether the cancellation has taken effect yet - and a dictionary
        with metadata.

    """
    response = await self._cancel_workflow_run(
        owner=owner, repository=repository, run_id=run_id, force=force, **kwargs
    )
    data, status_code = await process_async_response(response, default={})
    return cast(dict[str, Any], data), {"status_code": status_code}

gitea.actions.async_run_management.AsyncRunManagement.approve_workflow_run async

approve_workflow_run(
    owner: str, repository: str, run_id: int, **kwargs: Any
) -> tuple[dict[str, Any], dict[str, Any]]

Approve a workflow run that is waiting for approval.

A run triggered by a first-time contributor's pull request, or one whose jobs target an environment with a protection rule, sits in blocked until someone with write access approves it. This is that approval; a run that was not waiting for one answers 409.

Parameters:

Name	Type	Description	Default
owner	str	The owner of the repository.	required
repository	str	The name of the repository.	required
run_id	int	The ID of the run.	required
**kwargs	Any	Additional arguments for the request.	{}

Returns:

Type	Description
dict[str, Any]	A tuple containing the run as it now stands and a dictionary with
dict[str, Any]	metadata.

Source code in src/gitea/actions/async_run_management.py

async def approve_workflow_run(
    self, owner: str, repository: str, run_id: int, **kwargs: Any
) -> tuple[dict[str, Any], dict[str, Any]]:
    """Approve a workflow run that is waiting for approval.

    A run triggered by a first-time contributor's pull request, or one whose
    jobs target an environment with a protection rule, sits in `blocked`
    until someone with write access approves it. This is that approval; a run
    that was not waiting for one answers `409`.

    Args:
        owner: The owner of the repository.
        repository: The name of the repository.
        run_id: The ID of the run.
        **kwargs: Additional arguments for the request.

    Returns:
        A tuple containing the run as it now stands and a dictionary with
        metadata.

    """
    response = await self._approve_workflow_run(owner=owner, repository=repository, run_id=run_id, **kwargs)
    data, status_code = await process_async_response(response, default={})
    return cast(dict[str, Any], data), {"status_code": status_code}

gitea.actions.async_run_management.AsyncRunManagement.rerun_workflow_run async

rerun_workflow_run(
    owner: str,
    repository: str,
    run_id: int,
    failed_jobs_only: bool = False,
    **kwargs: Any,
) -> tuple[dict[str, Any], dict[str, Any]]

Rerun a workflow run.

Parameters:

Name	Type	Description	Default
owner	str	The owner of the repository.	required
repository	str	The name of the repository.	required
run_id	int	The ID of the run.	required
failed_jobs_only	bool	Whether to rerun only the jobs that failed, which is a different endpoint and the one to reach for when a run failed on one flaky job out of many.	False
**kwargs	Any	Additional arguments for the request.	{}

Returns:

Type	Description
dict[str, Any]	A tuple containing the run the rerun is on and a dictionary with
dict[str, Any]	metadata. The failed-jobs form answers 201 with no body, so an
tuple[dict[str, Any], dict[str, Any]]	empty payload with a 201 status is a rerun that started rather than
tuple[dict[str, Any], dict[str, Any]]	one that failed.

Source code in src/gitea/actions/async_run_management.py

async def rerun_workflow_run(
    self, owner: str, repository: str, run_id: int, failed_jobs_only: bool = False, **kwargs: Any
) -> tuple[dict[str, Any], dict[str, Any]]:
    """Rerun a workflow run.

    Args:
        owner: The owner of the repository.
        repository: The name of the repository.
        run_id: The ID of the run.
        failed_jobs_only: Whether to rerun only the jobs that failed, which
            is a different endpoint and the one to reach for when a run
            failed on one flaky job out of many.
        **kwargs: Additional arguments for the request.

    Returns:
        A tuple containing the run the rerun is on and a dictionary with
        metadata. The failed-jobs form answers `201` with no body, so an
        empty payload with a `201` status is a rerun that started rather than
        one that failed.

    """
    response = await self._rerun_workflow_run(
        owner=owner, repository=repository, run_id=run_id, failed_jobs_only=failed_jobs_only, **kwargs
    )
    data, status_code = await process_async_response(response, default={})
    return cast(dict[str, Any], data), {"status_code": status_code}

gitea.actions.async_run_management.AsyncRunManagement.rerun_workflow_job async

rerun_workflow_job(
    owner: str,
    repository: str,
    run_id: int,
    job_id: int,
    **kwargs: Any,
) -> tuple[dict[str, Any], dict[str, Any]]

Rerun one job of a workflow run.

This is the one job endpoint that takes the run as well: reading a job takes the job alone, because Gitea addresses one directly, but the rerun goes through the run it belongs to.

Parameters:

Name	Type	Description	Default
owner	str	The owner of the repository.	required
repository	str	The name of the repository.	required
run_id	int	The ID of the run the job belongs to.	required
job_id	int	The ID of the job.	required
**kwargs	Any	Additional arguments for the request.	{}

Returns:

Type	Description
dict[str, Any]	A tuple containing the job the rerun is on and a dictionary with
dict[str, Any]	metadata.

Source code in src/gitea/actions/async_run_management.py

async def rerun_workflow_job(
    self, owner: str, repository: str, run_id: int, job_id: int, **kwargs: Any
) -> tuple[dict[str, Any], dict[str, Any]]:
    """Rerun one job of a workflow run.

    This is the one job endpoint that takes the run as well: reading a job
    takes the job alone, because Gitea addresses one directly, but the rerun
    goes through the run it belongs to.

    Args:
        owner: The owner of the repository.
        repository: The name of the repository.
        run_id: The ID of the run the job belongs to.
        job_id: The ID of the job.
        **kwargs: Additional arguments for the request.

    Returns:
        A tuple containing the job the rerun is on and a dictionary with
        metadata.

    """
    response = await self._rerun_workflow_job(
        owner=owner, repository=repository, run_id=run_id, job_id=job_id, **kwargs
    )
    data, status_code = await process_async_response(response, default={})
    return cast(dict[str, Any], data), {"status_code": status_code}

gitea.actions.async_run_management.AsyncRunManagement.delete_workflow_run async

delete_workflow_run(
    owner: str, repository: str, run_id: int, **kwargs: Any
) -> tuple[dict[str, Any], dict[str, Any]]

Delete a workflow run, its jobs, its logs and its artifacts.

A run that has not finished cannot be deleted; cancel it first. What is deleted goes with it, so this is how a repository is cleared of the logs and artifacts of a run rather than only of the run's entry.

Parameters:

Name	Type	Description	Default
owner	str	The owner of the repository.	required
repository	str	The name of the repository.	required
run_id	int	The ID of the run.	required
**kwargs	Any	Additional arguments for the request.	{}

Returns:

Type	Description
dict[str, Any]	A tuple containing an empty dictionary - the endpoint answers 204
dict[str, Any]	with no body - and a dictionary with metadata.

Source code in src/gitea/actions/async_run_management.py

async def delete_workflow_run(
    self, owner: str, repository: str, run_id: int, **kwargs: Any
) -> tuple[dict[str, Any], dict[str, Any]]:
    """Delete a workflow run, its jobs, its logs and its artifacts.

    A run that has not finished cannot be deleted; cancel it first. What is
    deleted goes with it, so this is how a repository is cleared of the logs
    and artifacts of a run rather than only of the run's entry.

    Args:
        owner: The owner of the repository.
        repository: The name of the repository.
        run_id: The ID of the run.
        **kwargs: Additional arguments for the request.

    Returns:
        A tuple containing an empty dictionary - the endpoint answers `204`
        with no body - and a dictionary with metadata.

    """
    response = await self._delete_workflow_run(owner=owner, repository=repository, run_id=run_id, **kwargs)
    data, status_code = await process_async_response(response, default={})
    return cast(dict[str, Any], data), {"status_code": status_code}

Functions: