mainframe.endpoints.package

Attributes

Functions

submit_results(], auth, Depends]) → None)

lookup_package_info(], since, name, version, page, ...)

Lookup information on scanned packages based on name, version, or time scanned.

lookup_reported_packages(], since, name, page, size)

Lookup reported packages ordered by report time.

_deduplicate_packages(→ set[tuple[str, str]])

_get_packages_metadata(...)

batch_queue_package(], auth, Depends], 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[source]
mainframe.endpoints.package.logger: structlog.stdlib.BoundLogger[source]
mainframe.endpoints.package.DEFAULT_REPORTED_PACKAGE_PAGE = 1[source]
mainframe.endpoints.package.DEFAULT_REPORTED_PACKAGE_SIZE = 50[source]
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)]) None[source]
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, *, include_opengrep: bool = False) fastapi_pagination.Page[mainframe.models.schemas.Package] | collections.abc.Sequence[mainframe.models.schemas.Package][source]

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.

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))

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.

  • page – The page number of the result set.

  • size – The page size of the result set.

  • include_opengrep – Include stored evidence for an exact name and version, without claiming publication.

mainframe.endpoints.package.lookup_reported_packages(session: Annotated[sqlalchemy.orm.Session, Depends(get_db)], since: int | None = None, name: str | None = None, page: int = DEFAULT_REPORTED_PACKAGE_PAGE, size: int = DEFAULT_REPORTED_PACKAGE_SIZE) fastapi_pagination.Page[mainframe.models.schemas.Package][source]

Lookup reported packages ordered by report time.

mainframe.endpoints.package._deduplicate_packages(packages: list[mainframe.models.schemas.PackageSpecifier], session: sqlalchemy.orm.Session) set[tuple[str, str]][source]
mainframe.endpoints.package._get_packages_metadata(pypi_client: mainframe.pypi.PyPIClient, packages_to_check: set[tuple[str, str]]) collections.abc.Iterable[mainframe.pypi.PackageMetadata][source]
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[mainframe.pypi.PyPIClient, Depends(get_pypi_client)]) None[source]
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[mainframe.pypi.PyPIClient, Depends(get_pypi_client)]) mainframe.models.schemas.QueuePackageResponse[source]

Queue a package to be scanned when the next runner is available.

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