mainframe.endpoints.package
===========================

.. py:module:: mainframe.endpoints.package


Attributes
----------

.. autoapisummary::

   mainframe.endpoints.package.router
   mainframe.endpoints.package.logger
   mainframe.endpoints.package.DEFAULT_REPORTED_PACKAGE_PAGE
   mainframe.endpoints.package.DEFAULT_REPORTED_PACKAGE_SIZE


Functions
---------

.. autoapisummary::

   mainframe.endpoints.package.submit_results
   mainframe.endpoints.package.lookup_package_info
   mainframe.endpoints.package.lookup_reported_packages
   mainframe.endpoints.package._deduplicate_packages
   mainframe.endpoints.package._get_packages_metadata
   mainframe.endpoints.package.batch_queue_package
   mainframe.endpoints.package.queue_package


Module Contents
---------------

.. py:data:: router

.. py:data:: logger
   :type:  structlog.stdlib.BoundLogger

.. py:data:: DEFAULT_REPORTED_PACKAGE_PAGE
   :value: 1


.. py:data:: DEFAULT_REPORTED_PACKAGE_SIZE
   :value: 50


.. py:function:: 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

.. py:function:: 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]

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

   :param since: A int representing a Unix timestamp representing when to begin the search from.
   :param name: The name of the package.
   :param version: The version of the package.
   :param page: The page number of the result set.
   :param size: The page size of the result set.
   :param include_opengrep: Include stored evidence for an exact name and version, without claiming publication.


.. py:function:: 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]

   Lookup reported packages ordered by report time.


.. py:function:: _deduplicate_packages(packages: list[mainframe.models.schemas.PackageSpecifier], session: sqlalchemy.orm.Session) -> set[tuple[str, str]]

.. py:function:: _get_packages_metadata(pypi_client: mainframe.pypi.PyPIClient, packages_to_check: set[tuple[str, str]]) -> collections.abc.Iterable[mainframe.pypi.PackageMetadata | mainframe.models.schemas.PackageSpecifier]

.. py:function:: 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

.. py:function:: 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

   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.
   :rtype: 404