app.services.poll_plan

Value types and pure planning steps of a poll run.

Used by app.services.polling: PollPlan collects what a run found (counters, problems, documents to write, tokens to refresh, rows to delete); plan_project() applies the ownership rule, skips unchanged documents and plans deletions for one project listing; bounded() runs adapter calls with bounded concurrency.

Public API: PollProblem, PollCounts, PollOutcome, PollPlan, StoredState, bounded(), plan_project().

class app.services.poll_plan.PollProblem(*, code, message, project=None, path=None)

Bases: object

One finding of a poll run.

Parameters:

• code (str) – Stable code: source_error, projects_incomplete, listing_failed, listing_incomplete, fetch_failed, ownership_conflict, not_pollable, internal_error.

• message (str) – Actionable, secret-free description.

• project (str | None) – Affected project, if any.

• path (str | None) – Affected file path, if any.

code: str

message: str

project: str | None

path: str | None

to_json()

Convert to the JSON stored in intake_run.errors.

Return type:

dict[str, JsonValue]

Returns:

The mapping (None fields omitted).

class app.services.poll_plan.PollCounts(projects=0, documents=0, unchanged=0, fetched=0, created=0, updated=0, same_content=0, tokens_refreshed=0, deleted=0, skipped_owned=0, failed=0)

Bases: object

Counters of one poll run.

Parameters:

• projects (int) – Projects listed.

• documents (int) – Documents listed and owned by the registration.

• unchanged (int) – Documents skipped without fetching.

• fetched (int) – Documents fetched.

• created (int) – New published rows.

• updated (int) – Rows whose digest changed.

• same_content (int) – Fetched rows whose digest did not change.

• tokens_refreshed (int) – Unchanged rows whose change token moved.

• deleted (int) – Published rows removed by reconciliation.

• skipped_owned (int) – Files skipped (owned by an earlier registration).

• failed (int) – Listings or fetches that failed.

projects: int

documents: int

unchanged: int

fetched: int

created: int

updated: int

same_content: int

tokens_refreshed: int

deleted: int

skipped_owned: int

failed: int

to_json()

Convert to the JSON stored in intake_run.counts.

Return type:

dict[str, JsonValue]

Returns:

The counters.

class app.services.poll_plan.PollOutcome(*, source_id, status, run_id, counts, problems)

Bases: object

Result of polling one registration.

Parameters:

• source_id (str) – Registration id.

• status (PollStatus) – Outcome (disabled when not pollable).

• run_id (UUID | None) – Intake run id (None when nothing ran).

• counts (PollCounts) – Counters.

• problems (tuple[PollProblem, ...]) – Findings.

source_id: str

status: PollStatus

run_id: UUID | None

counts: PollCounts

problems: tuple[PollProblem, ...]

class app.services.poll_plan.PollPlan(counts=PollCounts(), problems=list(), complete=True, prepared=list(), refresh=list(), delete=list())

Bases: object

Everything collected for one registration, applied under lock.

Parameters:

• counts (PollCounts) – Counters.

• problems (list[PollProblem]) – Findings.

• complete (bool) – False once anything was incomplete or failed.

• prepared (list[PreparedDocument]) – Fetched documents to ingest.

• refresh (list[tuple[UUID, str | None]]) – (record id, new token) of unchanged documents.

• delete (list[UUID]) – Published row ids to delete.

counts: PollCounts

problems: list[PollProblem]

complete: bool

prepared: list[PreparedDocument]

refresh: list[tuple[UUID, str | None]]

delete: list[UUID]

fail(problem)

Record a problem that makes the run partial.

Parameters:

problem (PollProblem) – The problem.

Return type:

None

class app.services.poll_plan.StoredState(own, owned_elsewhere)

Bases: object

Stored state read before collecting.

Parameters:

• own (dict[str, dict[str, ChangeTokenEntry]]) – Project -> path -> stored entry of this registration.

• owned_elsewhere (frozenset[tuple[str, str]]) – (project, path) stored by earlier registrations.

own: dict[str, dict[str, ChangeTokenEntry]]

owned_elsewhere: frozenset[tuple[str, str]]

async app.services.poll_plan.bounded[T](limit, jobs)

Run coroutine factories with bounded concurrency, keeping order.

Parameters:

• limit (int) – Maximum concurrently running jobs.

• jobs (Iterable[Callable[[], Awaitable[T]]]) – Coroutine factories.

Return type:

list[T]

Returns:

Results in job order.

app.services.poll_plan.plan_project(project, listing, stored, plan)

Apply ownership, skip unchanged and plan deletions of a project.

Parameters:

• project (SourceProject) – The project.

• listing (DocumentListing) – Its document listing.

• stored (StoredState) – Stored state.

• plan (PollPlan) – Plan to fill.

Return type:

list[SourceDocument]

Returns:

Documents to fetch.