mainframe.models.schemas¶

Attributes¶

`RuleName`
`OpenGrepText`
`_DUPLICATE_RULES_ERROR`
`ReuseCount`
`ReuseBytes`
`ReuseMicros`

Classes¶

`ServerMetadata`	Server metadata.
`QueueStatus`	Cached summary of the package scanning queue.
`PerformanceStatus`	Database-derived rule and package performance totals.
`RulePerformance`	Internal per-rule performance totals.
`PublicStatistics`	Public package totals safe for unauthenticated consumers.
`AlertingConfigurationResponse`	Persisted production alerting configuration.
`AlertingConfigurationUpdate`	Mutable production alerting configuration.
`SuppressionRules`	Rule corpus for a suppression; null means every rule.
`SuppressionCreate`	A new suppression. Omitting rules suppresses every rule.
`SuppressionUpdate`	Replacement rule corpus for an existing suppression.
`SuppressionResponse`	A durable package-version suppression.
`SuppressionDeleteResponse`	Number of suppressions deleted by an operation.
`Error`	Error.
`OpenGrepFinding`	One bounded source-level finding produced by OpenGrep.
`OpenGrepDetails`	Read-only evidence and scan state, independent of Discord publication.
`Package`	Model representing a package queried from the database.
`PackageSpecifier`	Model used to specify a package by name and version.
`ReportPackageBody`	Model used to specify a package by name and version.
`ObservationKind`	Create a collection of name/value pairs.
`ObservationReport`	Model for a report using the PyPI Observation API.
`ScannerReuseMetrics`	Bounded per-job counters; exported to Prometheus without database storage.
`PackageScanResult`	Client payload to server containing the results of a package scan.
`PackageScanResultFail`	The client's reason as to why scanning a package failed.
`JobResult`	Package information of a requested job.
`GetRules`	!!! abstract "Usage Documentation"
`OpenGrepScanResult`	Complete or partial OpenGrep shadow result with preserved findings.
`OpenGrepScanResultFail`	Failed OpenGrep shadow result.
`OpenGrepAlert`	An alert delivered to Discord and eligible for OpenGrep processing.
`OpenGrepResult`	Completed OpenGrep shadow work awaiting bot publication.
`OpenGrepPublicationClaim`	Identify the publisher holding a shadow-result lease.
`OpenGrepPublicationProgress`	Durable Discord publication progress for retry-safe resumption.
`OpenGrepPublished`	Publication acknowledgement for a completed shadow result.
`NoJob`	Returned when no available jobs were found.
`QueuePackageResponse`	Returned after queueing a package. Contains the UUID.
`StatsResponse`	Recent system statistics.

Module Contents¶

mainframe.models.schemas.RuleName[source]¶

mainframe.models.schemas.OpenGrepText[source]¶

mainframe.models.schemas._DUPLICATE_RULES_ERROR = 'rules must not contain duplicates'[source]¶

class mainframe.models.schemas.ServerMetadata(/, **data: Any)[source]¶

Bases: pydantic.BaseModel

Server metadata.

server_commit: str[source]¶

rules_commit: str[source]¶

class mainframe.models.schemas.QueueStatus(/, **data: Any)[source]¶

Bases: pydantic.BaseModel

Cached summary of the package scanning queue.

queued: int[source]¶

in_progress: int[source]¶

retryable: int[source]¶

exhausted: int[source]¶

stranded: int[source]¶

total_backlog: int[source]¶

oldest_queued_at: datetime.datetime | None[source]¶

oldest_age_seconds: int | None[source]¶

sampled_at: datetime.datetime[source]¶

serialize_timestamp(value: datetime.datetime | None) → int | None[source]¶

class mainframe.models.schemas.PerformanceStatus(/, **data: Any)[source]¶

Bases: pydantic.BaseModel

Database-derived rule and package performance totals.

packages_scanned: int[source]¶

packages_failed: int[source]¶

packages_dead_lettered: int[source]¶

packages_above_production_threshold: int[source]¶

packages_reported: int[source]¶

production_score_threshold: int[source]¶

rule_hits: dict[str, int][source]¶

sampled_at: datetime.datetime[source]¶

class mainframe.models.schemas.RulePerformance(/, **data: Any)[source]¶

Bases: pydantic.BaseModel

Internal per-rule performance totals.

hits: dict[str, int][source]¶

sampled_at: datetime.datetime[source]¶

class mainframe.models.schemas.PublicStatistics(/, **data: Any)[source]¶

Bases: pydantic.BaseModel

Public package totals safe for unauthenticated consumers.

packages_scanned: int[source]¶

packages_reported: int[source]¶

sampled_at: datetime.datetime[source]¶

class mainframe.models.schemas.AlertingConfigurationResponse(/, **data: Any)[source]¶

Bases: pydantic.BaseModel

Persisted production alerting configuration.

production_score_threshold: int[source]¶

updated_at: datetime.datetime[source]¶

updated_by: str[source]¶

class mainframe.models.schemas.AlertingConfigurationUpdate(/, **data: Any)[source]¶

Bases: pydantic.BaseModel

Mutable production alerting configuration.

production_score_threshold: int = None[source]¶

class mainframe.models.schemas.SuppressionRules(/, **data: Any)[source]¶

Bases: pydantic.BaseModel

Rule corpus for a suppression; null means every rule.

rules: list[RuleName] | None[source]¶

classmethod rules_must_be_unique(rules: list[str] | None) → list[str] | None[source]¶

class mainframe.models.schemas.SuppressionCreate(/, **data: Any)[source]¶

Bases: SuppressionRules

A new suppression. Omitting rules suppresses every rule.

rules: list[RuleName] | None = None[source]¶

class mainframe.models.schemas.SuppressionUpdate(/, **data: Any)[source]¶

Bases: SuppressionRules

Replacement rule corpus for an existing suppression.

class mainframe.models.schemas.SuppressionResponse(/, **data: Any)[source]¶

Bases: pydantic.BaseModel

A durable package-version suppression.

suppression_id: uuid.UUID[source]¶

package_name: str[source]¶

package_version: str[source]¶

rules: list[str] | None[source]¶

created_at: datetime.datetime[source]¶

created_by: str[source]¶

updated_at: datetime.datetime[source]¶

updated_by: str[source]¶

classmethod from_db(suppression: mainframe.models.orm.Suppression) → Self[source]¶

class mainframe.models.schemas.SuppressionDeleteResponse(/, **data: Any)[source]¶

Bases: pydantic.BaseModel

Number of suppressions deleted by an operation.

deleted: int[source]¶

class mainframe.models.schemas.Error(/, **data: Any)[source]¶

Bases: pydantic.BaseModel

Error.

detail: str[source]¶

class mainframe.models.schemas.OpenGrepFinding(/, **data: Any)[source]¶

Bases: pydantic.BaseModel

One bounded source-level finding produced by OpenGrep.

rule_id: Annotated[str, Field(min_length=1, max_length=200)][source]¶

path: OpenGrepText[source]¶

start_line: int = None[source]¶

end_line: int = None[source]¶

message: OpenGrepText[source]¶

severity: Annotated[str, Field(min_length=1, max_length=20)][source]¶

evidence: Annotated[str, Field(min_length=1, max_length=32)][source]¶

confidence: Annotated[str, Field(min_length=1, max_length=20)][source]¶

execution_context: Annotated[str, Field(min_length=1, max_length=64)][source]¶

inspector_url: Annotated[str, Field(min_length=1, max_length=2048)][source]¶

class mainframe.models.schemas.OpenGrepDetails(/, **data: Any)[source]¶

Bases: pydantic.BaseModel

Read-only evidence and scan state, independent of Discord publication.

status: str[source]¶

commit: str | None[source]¶

duration_ms: int | None[source]¶

findings: list[OpenGrepFinding][source]¶

fail_reason: str | None[source]¶

finished_at: datetime.datetime | None[source]¶

class mainframe.models.schemas.Package(/, **data: Any)[source]¶

Bases: pydantic.BaseModel

Model representing a package queried from the database.

scan_id: str[source]¶

name: str[source]¶

version: str | None[source]¶

status: str | None[source]¶

score: int | None[source]¶

inspector_url: str | None[source]¶

rules: list[str] = [][source]¶

download_urls: list[str] = [][source]¶

queued_at: datetime.datetime | None[source]¶

queued_by: str | None[source]¶

reported_at: datetime.datetime | None[source]¶

reported_by: str | None[source]¶

report_summary: str | None[source]¶

pending_at: datetime.datetime | None[source]¶

pending_by: str | None[source]¶

attempt_count: int[source]¶

dead_lettered_at: datetime.datetime | None[source]¶

finished_at: datetime.datetime | None[source]¶

finished_by: str | None[source]¶

commit_hash: str | None[source]¶

opengrep: OpenGrepDetails | None = None[source]¶

classmethod from_db(scan: mainframe.models.orm.Scan, opengrep: mainframe.models.orm.OpenGrepScan | None = None) → Self[source]¶

serialize_dt(dt: datetime.datetime | None) → int | None[source]¶

class mainframe.models.schemas.PackageSpecifier(/, **data: Any)[source]¶

Bases: pydantic.BaseModel

Model used to specify a package by name and version.

name: A str of the name of the package to be scanned version: A str of the package version to scan.

model_config[source]¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str[source]¶

version: str[source]¶

class mainframe.models.schemas.ReportPackageBody(/, **data: Any)[source]¶

Bases: PackageSpecifier

Model used to specify a package by name and version.

name: A str of the name of the package to be scanned version: A str of the package version to scan.

inspector_url: str | None[source]¶

additional_information: str[source]¶

class mainframe.models.schemas.ObservationKind(*args, **kwds)[source]¶

Bases: enum.Enum

Create a collection of name/value pairs.

Example enumeration:

>>> class Color(Enum):
...     RED = 1
...     BLUE = 2
...     GREEN = 3

Access them by:

attribute access:
```
>>> Color.RED
<Color.RED: 1>
```
value lookup:
```
>>> Color(1)
<Color.RED: 1>
```
name lookup:
```
>>> Color['RED']
<Color.RED: 1>
```

Enumerations can be iterated over, and know how many members they have:

>>> len(Color)
3

>>> list(Color)
[<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

DependencyConfusion = 'is_dependency_confusion'[source]¶

Malware = 'is_malware'[source]¶

Spam = 'is_spam'[source]¶

Other = 'something_else'[source]¶

class mainframe.models.schemas.ObservationReport(/, **data: Any)[source]¶

Bases: pydantic.BaseModel

Model for a report using the PyPI Observation API.

kind: ObservationKind[source]¶

summary: str[source]¶

inspector_url: str | None[source]¶

extra: dict[str, Any] = None[source]¶

mainframe.models.schemas.ReuseCount[source]¶

mainframe.models.schemas.ReuseBytes[source]¶

mainframe.models.schemas.ReuseMicros[source]¶

class mainframe.models.schemas.ScannerReuseMetrics(/, **data: Any)[source]¶

Bases: pydantic.BaseModel

Bounded per-job counters; exported to Prometheus without database storage.

mode: Literal['off', 'observe', 'reuse'][source]¶

lookups: ReuseCount[source]¶

candidate_files: ReuseCount[source]¶

reused_files: ReuseCount[source]¶

reused_bytes: ReuseBytes[source]¶

inserted_files: ReuseCount[source]¶

evicted_files: ReuseCount[source]¶

errors: ReuseCount[source]¶

validated_files: ReuseCount[source]¶

mismatched_files: ReuseCount[source]¶

overhead_us: ReuseMicros[source]¶

engine_us: ReuseMicros[source]¶

engine_files: ReuseCount[source]¶

engine_bytes: ReuseBytes[source]¶

class mainframe.models.schemas.PackageScanResult(/, **data: Any)[source]¶

Bases: PackageSpecifier

Client payload to server containing the results of a package scan.

scan_reuse: ScannerReuseMetrics | None = None[source]¶

commit: str[source]¶

score: int = 0[source]¶

inspector_url: str | None = None[source]¶

rules_matched: list[str] = [][source]¶

attempt: int | None = None[source]¶

assignment_id: uuid.UUID | None = None[source]¶

class mainframe.models.schemas.PackageScanResultFail(/, **data: Any)[source]¶

Bases: PackageSpecifier

The client’s reason as to why scanning a package failed.

scan_reuse: ScannerReuseMetrics | None = None[source]¶

reason: str[source]¶

attempt: int | None = None[source]¶

assignment_id: uuid.UUID | None = None[source]¶

class mainframe.models.schemas.JobResult(/, **data: Any)[source]¶

Bases: pydantic.BaseModel

Package information of a requested job.

name: str[source]¶

version: str[source]¶

distributions: list[str][source]¶

hash: str[source]¶

attempt: int = None[source]¶

assignment_id: uuid.UUID[source]¶

class mainframe.models.schemas.GetRules(/, **data: Any)[source]¶

Bases: pydantic.BaseModel

!!! abstract “Usage Documentation”: [Models](../concepts/models.md)

A base class for creating Pydantic models.

__class_vars__[source]¶: The names of the class variables defined on the model.

__private_attributes__[source]¶: Metadata about the private attributes of the model.

__signature__[source]¶: The synthesized __init__ [Signature][inspect.Signature] of the model.

__pydantic_complete__[source]¶: Whether model building is completed, or if there are still undefined fields.

__pydantic_core_schema__[source]¶: The core schema of the model.

__pydantic_custom_init__[source]¶: Whether the model has a custom __init__ function.

__pydantic_decorators__[source]¶: Metadata containing the decorators defined on the model. This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.

__pydantic_generic_metadata__[source]¶: A dictionary containing metadata about generic Pydantic models. The origin and args items map to the [__origin__][genericalias.__origin__] and [__args__][genericalias.__args__] attributes of [generic aliases][types-genericalias], and the parameter item maps to the __parameter__ attribute of generic classes.

__pydantic_parent_namespace__[source]¶: Parent namespace of the model, used for automatic rebuilding of models.

__pydantic_post_init__[source]¶: The name of the post-init method for the model, if defined.

__pydantic_root_model__[source]¶: Whether the model is a [RootModel][pydantic.root_model.RootModel].

__pydantic_serializer__[source]¶: The pydantic-core SchemaSerializer used to dump instances of the model.

__pydantic_validator__[source]¶: The pydantic-core SchemaValidator used to validate instances of the model.

__pydantic_fields__[source]¶: A dictionary of field names and their corresponding [FieldInfo][pydantic.fields.FieldInfo] objects.

__pydantic_computed_fields__[source]¶: A dictionary of computed field names and their corresponding [ComputedFieldInfo][pydantic.fields.ComputedFieldInfo] objects.