`prefect.server.services.task_run_recorder`

Functions

`task_run_from_event`

task_run_from_event(event: ReceivedEvent) -> TaskRun

`db_recordable_task_run_from_event`

db_recordable_task_run_from_event(event: ReceivedEvent) -> tuple[TaskRun, dict[str, Any]]

`record_task_run_event`

record_task_run_event(event: ReceivedEvent, depth: int = 0) -> None

Record a single task run event in the database. Delegates to record_bulk_task_run_events, which already retries once on IntegrityError to recover from TOCTOU races against concurrent recorders. Any IntegrityError that survives the retry is treated as an unrecoverable duplicate and the event is discarded.

`record_bulk_task_run_events`

record_bulk_task_run_events(events: list[ReceivedEvent]) -> None

Record multiple task run events in the database, taking advantage of bulk inserts. Retries once on IntegrityError to handle TOCTOU races between concurrent recorder instances: when two batches reference the same task_run.id with different natural keys, one batch’s existence-check SELECT may run before the other batch’s INSERT commits. The retry re-runs the SELECT in a fresh session so the conflict target is chosen against the now-visible row.

`consumer`

consumer(write_batch_size: int, flush_every: int, max_persist_retries: int = DEFAULT_PERSIST_MAX_RETRIES) -> AsyncGenerator[MessageHandler, None]

Classes

`RetryableEvent`

`TaskRunRecorder`

Constructs task runs and states from client-emitted events Methods:

`all_services`

all_services(cls) -> Sequence[type[Self]]

Get list of all service classes

`enabled`

enabled(cls) -> bool

Whether the service is enabled

`enabled_services`

enabled_services(cls) -> list[type[Self]]

Get list of enabled service classes

`environment_variable_name`

environment_variable_name(cls) -> str

`run_services`

run_services(cls) -> NoReturn

Run enabled services until cancelled.

`running`

running(cls) -> AsyncGenerator[None, None]

A context manager that runs enabled services on entry and stops them on exit.

`service_settings`

service_settings(cls) -> ServicesBaseSetting

`service_settings`

service_settings(cls) -> ServicesBaseSetting

The Prefect setting that controls whether the service is enabled

`start`

start(self, max_persist_retries: int = DEFAULT_PERSIST_MAX_RETRIES) -> NoReturn

`start`

start(self) -> NoReturn

Start running the service, which may run indefinitely

`started_event`

started_event(self) -> asyncio.Event

`started_event`

started_event(self, value: asyncio.Event) -> None

`stop`

stop(self) -> None

`stop`

stop(self) -> None

Stop the service

​prefect.server.services.task_run_recorder

​Functions

​task_run_from_event

​db_recordable_task_run_from_event

​record_task_run_event

​record_bulk_task_run_events

​consumer

​Classes

​RetryableEvent

​TaskRunRecorder

​all_services

​enabled

​enabled_services

​environment_variable_name

​run_services

​running

​service_settings

​service_settings

​start

​start

​started_event

​started_event

​stop

​stop

`prefect.server.services.task_run_recorder`

Functions

`task_run_from_event`

`db_recordable_task_run_from_event`

`record_task_run_event`

`record_bulk_task_run_events`

`consumer`

Classes

`RetryableEvent`

`TaskRunRecorder`

`all_services`

`enabled`

`enabled_services`

`environment_variable_name`

`run_services`

`running`

`service_settings`

`service_settings`

`start`

`start`

`started_event`

`started_event`

`stop`

`stop`