mainframe.endpoints.package

Attributes

router
logger

Functions

submit_results(result, session, auth)
lookup_package_info(], since, name, version, page, size)	Lookup information on scanned packages based on name, version, or time
_deduplicate_packages(→ set[tuple[str, str]])
_get_packages_metadata(...)
batch_queue_package(packages, session, auth, pypi_client)
queue_package(], auth, Depends], pypi_client, ...)	Queue a package to be scanned when the next runner is available

Module Contents

mainframe.endpoints.package.router

mainframe.endpoints.package.logger: structlog.stdlib.BoundLogger

mainframe.endpoints.package.submit_results(result: mainframe.models.schemas.PackageScanResult | mainframe.models.schemas.PackageScanResultFail, session: Annotated[sqlalchemy.orm.Session, Depends(get_db)], auth: Annotated[mainframe.json_web_token.AuthenticationData, Depends(validate_token)])

mainframe.endpoints.package.lookup_package_info(session: Annotated[sqlalchemy.orm.Session, Depends(get_db)], since: int | None = None, name: str | None = None, version: str | None = None, page: int | None = None, size: int | None = None) → fastapi_pagination.Page[mainframe.models.schemas.Package] | collections.abc.Sequence[mainframe.models.schemas.Package]

Lookup information on scanned packages based on name, version, or time scanned. If multiple packages are returned, they are ordered with the most recently queued package first.

Parameters:

• since – A int representing a Unix timestamp representing when to begin the search from.

• name – The name of the package.

• version – The version of the package.

• session – DB session.

Only certain combinations of parameters are allowed. A query is valid if any of the following combinations are used:

• name and version: Return the package with name name and version version, if it exists.

• name and since: Find all packages with name name since since.

• since: Find all packages since since.

• name: Find all packages with name name.

All other combinations are disallowed.

In more formal terms, a query is valid

iff ((name and not since) or (not version and since))

where a given variable name means that query parameter was passed. Equivalently, a request is invalid

iff (not (name or since) or (version and since))

mainframe.endpoints.package._deduplicate_packages(packages: list[mainframe.models.schemas.PackageSpecifier], session: sqlalchemy.orm.Session) → set[tuple[str, str]]

mainframe.endpoints.package._get_packages_metadata(pypi_client: letsbuilda.pypi.PyPIServices, packages_to_check: set[tuple[str, str]]) → collections.abc.Iterable[letsbuilda.pypi.Package]

mainframe.endpoints.package.batch_queue_package(packages: list[mainframe.models.schemas.PackageSpecifier], session: Annotated[sqlalchemy.orm.Session, Depends(get_db)], auth: Annotated[mainframe.json_web_token.AuthenticationData, Depends(validate_token)], pypi_client: Annotated[letsbuilda.pypi.PyPIServices, Depends(get_pypi_client)])

mainframe.endpoints.package.queue_package(package: mainframe.models.schemas.PackageSpecifier, session: Annotated[sqlalchemy.orm.Session, Depends(get_db)], auth: Annotated[mainframe.json_web_token.AuthenticationData, Depends(validate_token)], pypi_client: Annotated[letsbuilda.pypi.PyPIServices, Depends(get_pypi_client)]) → mainframe.models.schemas.QueuePackageResponse

Queue a package to be scanned when the next runner is available :param Body: Request body paramters :param session: Database session :param pypi_client: client instance used to interact with PyPI JSON API

Returns:

The given package and version combination was not found on PyPI 409: The given package and version combination has already been queued

Return type:

404