API: client for Pulp 3

class pubtools.pulplib.Pulp3Client(url: str, auth: tuple | None = None, cert: str | Tuple[str, str] | None = None, verify: bool = True, timeout: float | None = None, max_retries: int = 5, retry_min_wait: float = 1.0, retry_max_wait: float = 60.0, domain: str | None = None, **kwargs)[source]

An async client for the Pulp 3.x API.

This class provides high-level async methods for querying a Pulp 3 server and performing actions in Pulp.

Usage:

This client is designed to be used as an async context manager with AnyIO’s Trio backend:

async with Pulp3Client(url="https://pulp.example.com/") as client:
    repo = await client.create_repository("test-repo")
    print(repo["name"])

Backend:

The client uses: - AnyIO with Trio backend for structured concurrency - httpx for async HTTP requests - Task groups for managing concurrent operations

Rate Limiting:

The client supports automatic rate limiting to avoid overwhelming the server.

Parameters:

  • url – URL of a Pulp 3 server, e.g. https://pulp.example.com/

  • auth – Optional tuple of (username, password) for authentication

  • cert – Optional path to client certificate file

  • verify – Whether to verify SSL certificates (default: True)

  • timeout – Request timeout in seconds (default: 120.0)

  • max_retries – Maximum retry attempts for failed requests (default: 5)

  • retry_min_wait – Minimum wait between retries in seconds (default: 1.0)

  • retry_max_wait – Maximum wait between retries in seconds (default: 60.0)

  • domain – Pulp domain name (default: from PULP_DOMAIN env or ‘default’)

  • **kwargs – Additional arguments passed to httpx.AsyncClient

async get_status() → Dict[str, Any][source]

Get Pulp server status.

Returns:

Dictionary containing server status information

async create_repository(name: str, **kwargs) → Dict[str, Any][source]

Create a new repository.

Parameters:

  • name – Repository name

  • **kwargs – Additional repository attributes (description, retain_repo_versions, etc.)

Returns:

Created repository data

async poll_task(task_href_or_id: str, poll_interval: float = 5.0, timeout: float = 3600.0) → Dict[str, Any][source]

Poll a task until completion.

Parameters:

  • task_href_or_id – Task href (e.g., ‘/api/pulp/…/tasks/abc/’) or task ID (e.g., ‘abc’)

  • poll_interval – Seconds between poll attempts (default: 5.0)

  • timeout – Maximum time to wait in seconds (default: 3600.0 / 1 hour)

Returns:

Completed task data dictionary

Raises:
async create_publication(repository_href: str, **kwargs) → str[source]

Create a publication for a repository.

Parameters:

  • repository_href – Repository href to publish

  • **kwargs – Additional publish options

Returns:

Task href string (e.g., ‘/api/pulp/…/tasks/abc/’)

Raises:

httpx.HTTPStatusError – If request fails

async modify_repo_content(repository_href_or_id: str, add_content_units: list | None = None, remove_content_units: list | None = None) → str[source]

Modify repository content by adding or removing content units.

Parameters:

  • repository_href_or_id – Repository href (e.g., ‘/api/…/repositories/rpm/rpm/abc/’) or repository ID (e.g., ‘abc’)

  • add_content_units – List of content unit hrefs to add (default: None)

  • remove_content_units – List of content unit hrefs to remove (default: None)

Returns:

Task href string (e.g., ‘/api/pulp/…/tasks/xyz/’)

Raises:

  • ValueError – If both add_content_units and remove_content_units are empty

  • httpx.HTTPStatusError – If request fails

async create_distribution(repository_href: str, base_path: str, name: str) → str[source]

Create a distribution for a repository.

Parameters:

  • repository_href – Repository href to add distribution

  • base_path – Base path for the distribution (URL path component)

  • name – Name of the distribution

Returns:

Task href string (e.g., ‘/api/pulp/…/tasks/xyz/’)

Raises:

httpx.HTTPStatusError – If request fails

build_query_sha256(checksums: list) → str[source]

Build a Pulp query for searching content by SHA256 checksums.

Parameters:

checksums – List of SHA256 checksum strings

Returns:

Query string suitable for search_content(), e.g., “(sha256=abc OR sha256=def)”

Example

>>> client.build_query_sha256(["abc123", "def456"])
'(sha256=abc123 OR sha256=def456)'

Note

This method will be refactored/reimplemented to support generic field queries in the future.

async search_content(query: str, fields: list | None = None, offset: int = 0, limit: int = 1000) → list[source]

Search for RPM package content.

Parameters:

  • query – Query string (e.g., from build_query_sha256)

  • fields – Optional list of field names to return (default: all fields)

  • offset – Offset for pagination (default: 0)

  • limit – Limit for pagination (default: 1000)

Returns:

List of matching content unit dictionaries

Raises:

httpx.HTTPStatusError – If request fails

Note

This method currently does not support pagination automatically. Use offset/limit in params for manual pagination.

Errors

class pubtools.pulplib.Pulp3Exception[source]
class pubtools.pulplib.Pulp3TaskException[source]