> ## 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. # flows # `prefect.server.models.flows` Functions for interacting with flow ORM objects. Intended for internal use by the Prefect REST API. ## Functions ### `create_flow` ```python theme={null} create_flow(db: PrefectDBInterface, session: AsyncSession, flow: schemas.core.Flow) -> orm_models.Flow ``` Creates a new flow. If a flow with the same name already exists, the existing flow is returned. **Args:** * `session`: a database session * `flow`: a flow model **Returns:** * orm\_models.Flow: the newly-created or existing flow ### `update_flow` ```python theme={null} update_flow(db: PrefectDBInterface, session: AsyncSession, flow_id: UUID, flow: schemas.actions.FlowUpdate) -> bool ``` Updates a flow. **Args:** * `session`: a database session * `flow_id`: the flow id to update * `flow`: a flow update model **Returns:** * whether or not matching rows were found to update ### `read_flow` ```python theme={null} read_flow(db: PrefectDBInterface, session: AsyncSession, flow_id: UUID) -> Optional[orm_models.Flow] ``` Reads a flow by id. **Args:** * `session`: A database session * `flow_id`: a flow id **Returns:** * orm\_models.Flow: the flow ### `read_flow_by_name` ```python theme={null} read_flow_by_name(db: PrefectDBInterface, session: AsyncSession, name: str) -> Optional[orm_models.Flow] ``` Reads a flow by name. **Args:** * `session`: A database session * `name`: a flow name **Returns:** * orm\_models.Flow: the flow ### `read_flows` ```python theme={null} read_flows(db: PrefectDBInterface, session: AsyncSession, flow_filter: Union[schemas.filters.FlowFilter, None] = None, flow_run_filter: Union[schemas.filters.FlowRunFilter, None] = None, task_run_filter: Union[schemas.filters.TaskRunFilter, None] = None, deployment_filter: Union[schemas.filters.DeploymentFilter, None] = None, work_pool_filter: Union[schemas.filters.WorkPoolFilter, None] = None, sort: schemas.sorting.FlowSort = schemas.sorting.FlowSort.NAME_ASC, offset: Union[int, None] = None, limit: Union[int, None] = None) -> Sequence[orm_models.Flow] ``` Read multiple flows. **Args:** * `session`: A database session * `flow_filter`: only select flows that match these filters * `flow_run_filter`: only select flows whose flow runs match these filters * `task_run_filter`: only select flows whose task runs match these filters * `deployment_filter`: only select flows whose deployments match these filters * `work_pool_filter`: only select flows whose work pools match these filters * `offset`: Query offset * `limit`: Query limit **Returns:** * List\[orm\_models.Flow]: flows ### `count_flows` ```python theme={null} count_flows(db: PrefectDBInterface, session: AsyncSession, flow_filter: Union[schemas.filters.FlowFilter, None] = None, flow_run_filter: Union[schemas.filters.FlowRunFilter, None] = None, task_run_filter: Union[schemas.filters.TaskRunFilter, None] = None, deployment_filter: Union[schemas.filters.DeploymentFilter, None] = None, work_pool_filter: Union[schemas.filters.WorkPoolFilter, None] = None) -> int ``` Count flows. **Args:** * `session`: A database session * `flow_filter`: only count flows that match these filters * `flow_run_filter`: only count flows whose flow runs match these filters * `task_run_filter`: only count flows whose task runs match these filters * `deployment_filter`: only count flows whose deployments match these filters * `work_pool_filter`: only count flows whose work pools match these filters **Returns:** * count of flows ### `delete_flow` ```python theme={null} delete_flow(db: PrefectDBInterface, session: AsyncSession, flow_id: UUID) -> bool ``` Delete a flow by id. **Args:** * `session`: A database session * `flow_id`: a flow id **Returns:** * whether or not the flow was deleted ### `delete_flows` ```python theme={null} delete_flows(db: PrefectDBInterface, session: AsyncSession, flow_ids: List[UUID]) -> List[UUID] ``` Delete multiple flows by their IDs. This also deletes all associated deployments (hard delete). **Args:** * `session`: A database session * `flow_ids`: a list of flow ids to delete **Returns:** * List\[UUID]: the IDs of the flows that were deleted ### `read_flow_labels` ```python theme={null} read_flow_labels(db: PrefectDBInterface, session: AsyncSession, flow_id: UUID) -> Union[schemas.core.KeyValueLabels, None] ``` Built with [Mintlify](https://mintlify.com).