app.services.intake_checks

Request validation and refusal checks of CI push intake.

Pure helpers used by app.services.intake (which owns the order of the checks, the run log and the write); see that module’s docstring for the full contract. Nothing here touches storage.

Public API: MAX_BODY_BYTES, MAX_CONTENT_BYTES, CandidateSubmission, auth_error(), peek_source_id(), decode_body(), authenticate(), named_source(), validate_submission(), check_registration(), build_document(), check_document(), issues_json().

app.services.intake_checks.MAX_BODY_BYTES = 8388608

Largest accepted request body (JSON escaping may inflate content).

app.services.intake_checks.MAX_CONTENT_BYTES = 1048576

Largest accepted Markdown document (UTF-8 bytes).

class app.services.intake_checks.CandidateSubmission(*, source_id, project, adr_uuid, ref, content, path=None, delivery_id=None, author=None)

Bases: BaseModel

Body of PUT/POST /api/v1/candidates; extra keys -> 422.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

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

source_id: str

project: str

adr_uuid: UUID

ref: str

content: str

path: str | None

delivery_id: str | None

author: str | None

app.services.intake_checks.auth_error(message, *, presented)

Build the 401 for a missing or unacceptable push token.

Parameters:

• message (str) – Reason (never contains the token).

• presented (bool) – Whether the request carried a token.

Return type:

AuthenticationError

Returns:

The error.

app.services.intake_checks.authenticate(sources, token, source_id)

Check the push token against the named registration.

Parameters:

• sources (Sequence[SourceRegistration]) – All configured registrations.

• token (str) – Presented bearer token.

• source_id (str | None) – Requested registration id, if the body names one.

Return type:

SourceRegistration | None

Returns:

The registration; None when the body names none but the token is valid for some registration.

Raises:

AuthenticationError – If the token is not acceptable.

app.services.intake_checks.build_document(source, submission)

Build the ingest input from a submission.

Parameters:

• source (SourceRegistration) – The registration.

• submission (CandidateSubmission) – The validated body.

Return type:

IncomingDocument

Returns:

The document.

Raises:

• PayloadTooLargeError – content_too_large.

• ValidationFailedError – bad_encoding.

app.services.intake_checks.check_document(prepared, submission)

Refuse unusable documents and UUID mismatches.

Parameters:

• prepared (PreparedDocument) – Parsed document.

• submission (CandidateSubmission) – The validated body.

Raises:

ValidationFailedError – invalid_document (with issues) or uuid_mismatch.

Return type:

None

app.services.intake_checks.check_registration(source, submission)

Check push intake, scope and ref against the registration.

Parameters:

• source (SourceRegistration) – The authenticated registration.

• submission (CandidateSubmission) – The validated body.

Raises:

• ForbiddenError – push_disabled / project_out_of_scope.

• ValidationFailedError – published_ref_not_allowed.

Return type:

None

app.services.intake_checks.decode_body(body)

Decode a JSON body.

Parameters:

body (bytes) – Raw request body.

Return type:

object

Returns:

The decoded value; None when it is not valid JSON.

app.services.intake_checks.issues_json(issues)

Serialise issues to JSON values.

Parameters:

issues (Sequence[Issue]) – Validation issues.

Return type:

list[JsonValue]

Returns:

One {severity, code, message} object per issue.

app.services.intake_checks.named_source(sources, token, source_id)

Check the push token against one registration.

Parameters:

• sources (Sequence[SourceRegistration]) – All configured registrations.

• token (str) – Presented bearer token.

• source_id (str) – Requested registration id.

Return type:

SourceRegistration

Returns:

The registration.

Raises:

AuthenticationError – If the registration is unknown or the token is not one of its push tokens (same answer for both).

app.services.intake_checks.peek_source_id(raw)

Read source_id from a decoded body, if it is a string.

Parameters:

raw (object) – Decoded JSON body.

Return type:

str | None

Returns:

The requested registration id, or None.

app.services.intake_checks.validate_submission(raw)

Validate the decoded body.

Parameters:

raw (object) – Decoded JSON (None when the body was not JSON).

Return type:

CandidateSubmission

Returns:

The submission.

Raises:

ValidationFailedError – invalid_request with errors.