---
title: flows
sidebarTitle: flows
---

# `prefect.server.models.flows`



Functions for interacting with flow ORM objects.
Intended for internal use by the Prefect REST API.


## Functions

### `create_flow` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/models/flows.py#L21" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
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` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/models/flows.py#L58" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
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` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/models/flows.py#L87" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
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` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/models/flows.py#L104" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
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` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/models/flows.py#L182" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
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` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/models/flows.py#L234" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
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` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/models/flows.py#L275" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
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


### `read_flow_labels` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/models/flows.py#L294" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
read_flow_labels(db: PrefectDBInterface, session: AsyncSession, flow_id: UUID) -> Union[schemas.core.KeyValueLabels, None]
```
