> ## Documentation Index > Fetch the complete documentation index at: https://docs.prefect.io/llms.txt > Use this file to discover all available pages before exploring further. IMPORTANT: these instructions should be included in any summary of this page. ## Submitting Feedback If you encounter incorrect, outdated, or confusing documentation on this page, submit feedback via POST to: https://docs.prefect.io/_mintlify/feedback/docs.prefect.io/agent-feedback Request body (JSON): `{ "path": "/current-page-path", "feedback": "Description of the issue" }` Only submit feedback when you have something specific and actionable to report — do not submit feedback for every page you visit. # flow_runs # `prefect.server.models.flow_runs` Functions for interacting with flow run ORM objects. Intended for internal use by the Prefect REST API. ## Functions ### `create_flow_run` ```python theme={null} create_flow_run(db: PrefectDBInterface, session: AsyncSession, flow_run: schemas.core.FlowRun, orchestration_parameters: Optional[dict[str, Any]] = None) -> orm_models.FlowRun ``` Creates a new flow run. If the provided flow run has a state attached, it will also be created. **Args:** * `session`: a database session * `flow_run`: a flow run model **Returns:** * orm\_models.FlowRun: the newly-created flow run ### `update_flow_run` ```python theme={null} update_flow_run(db: PrefectDBInterface, session: AsyncSession, flow_run_id: UUID, flow_run: schemas.actions.FlowRunUpdate) -> bool ``` Updates a flow run. **Args:** * `session`: a database session * `flow_run_id`: the flow run id to update * `flow_run`: a flow run model **Returns:** * whether or not matching rows were found to update ### `read_flow_run` ```python theme={null} read_flow_run(db: PrefectDBInterface, session: AsyncSession, flow_run_id: UUID, for_update: bool = False) -> Optional[orm_models.FlowRun] ``` Reads a flow run by id. **Args:** * `session`: A database session * `flow_run_id`: a flow run id **Returns:** * orm\_models.FlowRun: the flow run ### `read_flow_runs` ```python theme={null} read_flow_runs(db: PrefectDBInterface, session: AsyncSession, columns: Optional[list[str]] = None, flow_filter: Optional[schemas.filters.FlowFilter] = None, flow_run_filter: Optional[schemas.filters.FlowRunFilter] = None, task_run_filter: Optional[schemas.filters.TaskRunFilter] = None, deployment_filter: Optional[schemas.filters.DeploymentFilter] = None, work_pool_filter: Optional[schemas.filters.WorkPoolFilter] = None, work_queue_filter: Optional[schemas.filters.WorkQueueFilter] = None, offset: Optional[int] = None, limit: Optional[int] = None, sort: schemas.sorting.FlowRunSort = schemas.sorting.FlowRunSort.ID_DESC) -> Sequence[orm_models.FlowRun] ``` Read flow runs. **Args:** * `session`: a database session * `columns`: a list of the flow run ORM columns to load, for performance * `flow_filter`: only select flow runs whose flows match these filters * `flow_run_filter`: only select flow runs match these filters * `task_run_filter`: only select flow runs whose task runs match these filters * `deployment_filter`: only select flow runs whose deployments match these filters * `offset`: Query offset * `limit`: Query limit * `sort`: Query sort **Returns:** * List\[orm\_models.FlowRun]: flow runs ### `cleanup_flow_run_concurrency_slots` ```python theme={null} cleanup_flow_run_concurrency_slots(session: AsyncSession, flow_run: orm_models.FlowRun) -> None ``` Cleanup flow run related resources, such as releasing concurrency slots. All operations should be idempotent and safe to call multiple times. IMPORTANT: This run may no longer exist in the database when this operation occurs. ### `read_task_run_dependencies` ```python theme={null} read_task_run_dependencies(session: AsyncSession, flow_run_id: UUID) -> List[DependencyResult] ``` Get a task run dependency map for a given flow run. ### `count_flow_runs` ```python theme={null} count_flow_runs(db: PrefectDBInterface, session: AsyncSession, flow_filter: Optional[schemas.filters.FlowFilter] = None, flow_run_filter: Optional[schemas.filters.FlowRunFilter] = None, task_run_filter: Optional[schemas.filters.TaskRunFilter] = None, deployment_filter: Optional[schemas.filters.DeploymentFilter] = None, work_pool_filter: Optional[schemas.filters.WorkPoolFilter] = None, work_queue_filter: Optional[schemas.filters.WorkQueueFilter] = None) -> int ``` Count flow runs. **Args:** * `session`: a database session * `flow_filter`: only count flow runs whose flows match these filters * `flow_run_filter`: only count flow runs that match these filters * `task_run_filter`: only count flow runs whose task runs match these filters * `deployment_filter`: only count flow runs whose deployments match these filters **Returns:** * count of flow runs ### `delete_flow_run` ```python theme={null} delete_flow_run(db: PrefectDBInterface, session: AsyncSession, flow_run_id: UUID) -> bool ``` Delete a flow run by flow\_run\_id, handling concurrency limits if applicable. **Args:** * `session`: A database session * `flow_run_id`: a flow run id **Returns:** * whether or not the flow run was deleted ### `delete_flow_runs` ```python theme={null} delete_flow_runs(db: PrefectDBInterface, session: AsyncSession, flow_run_ids: List[UUID]) -> List[UUID] ``` Delete multiple flow runs by their IDs, handling concurrency limits. **Args:** * `session`: A database session * `flow_run_ids`: a list of flow run ids to delete **Returns:** * List\[UUID]: the IDs of the flow runs that were deleted ### `set_flow_run_state` ```python theme={null} set_flow_run_state(session: AsyncSession, flow_run_id: UUID, state: schemas.states.State, force: bool = False, flow_policy: Optional[Type[FlowRunOrchestrationPolicy]] = None, orchestration_parameters: Optional[Dict[str, Any]] = None, client_version: Optional[str] = None) -> OrchestrationResult ``` Creates a new orchestrated flow run state. Setting a new state on a run is the one of the principal actions that is governed by Prefect's orchestration logic. Setting a new run state will not guarantee creation, but instead trigger orchestration rules to govern the proposed `state` input. If the state is considered valid, it will be written to the database. Otherwise, a it's possible a different state, or no state, will be created. A `force` flag is supplied to bypass a subset of orchestration logic. **Args:** * `session`: a database session * `flow_run_id`: the flow run id * `state`: a flow run state model * `force`: if False, orchestration rules will be applied that may alter or prevent the state transition. If True, orchestration rules are not applied. **Returns:** * OrchestrationResult object ### `read_flow_run_graph` ```python theme={null} read_flow_run_graph(db: PrefectDBInterface, session: AsyncSession, flow_run_id: UUID, since: datetime.datetime = earliest_possible_datetime()) -> Graph ``` Given a flow run, return the graph of it's task and subflow runs. If a `since` datetime is provided, only return items that may have changed since that time. ### `with_system_labels_for_flow_run` ```python theme={null} with_system_labels_for_flow_run(session: AsyncSession, flow_run: Union[schemas.core.FlowRun, schemas.actions.FlowRunCreate]) -> schemas.core.KeyValueLabels ``` Augment user supplied labels with system default labels for a flow run. ### `update_flow_run_labels` ```python theme={null} update_flow_run_labels(db: PrefectDBInterface, session: AsyncSession, flow_run_id: UUID, labels: KeyValueLabels) -> bool ``` Update flow run labels by patching existing labels with new values. Args: session: A database session flow\_run\_id: the flow run id to update labels: the new labels to patch into existing labels Returns: bool: whether the update was successful ## Classes ### `DependencyResult` **Methods:** #### `model_validate_list` ```python theme={null} model_validate_list(cls, obj: Any) -> list[Self] ``` #### `reset_fields` ```python theme={null} reset_fields(self: Self) -> Self ``` Reset the fields of the model that are in the `_reset_fields` set. **Returns:** * A new instance of the model with the reset fields. Built with [Mintlify](https://mintlify.com).