Skip to content

prefect.server.api.flow_run_notification_policies

Routes for interacting with flow run notification policy objects.

create_flow_run_notification_policy(flow_run_notification_policy, db=Depends(provide_database_interface)) async

Creates a new flow run notification policy.

Source code in src/prefect/server/api/flow_run_notification_policies.py 
22
23
24
25
26
27
28
29
30
31
32
33
@router.post("/", status_code=status.HTTP_201_CREATED)
async def create_flow_run_notification_policy(
    flow_run_notification_policy: schemas.actions.FlowRunNotificationPolicyCreate,
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> schemas.core.FlowRunNotificationPolicy:
    """
    Creates a new flow run notification policy.
    """
    async with db.session_context(begin_transaction=True) as session:
        return await models.flow_run_notification_policies.create_flow_run_notification_policy(
            session=session, flow_run_notification_policy=flow_run_notification_policy
        )

delete_flow_run_notification_policy(flow_run_notification_policy_id=Path(..., description='The flow run notification policy id', alias='id'), db=Depends(provide_database_interface)) async

Delete a flow run notification policy by id.

Source code in src/prefect/server/api/flow_run_notification_policies.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_flow_run_notification_policy(
    flow_run_notification_policy_id: UUID = Path(
        ..., description="The flow run notification policy id", alias="id"
    ),
    db: PrefectDBInterface = Depends(provide_database_interface),
):
    """
    Delete a flow run notification policy by id.
    """
    async with db.session_context(begin_transaction=True) as session:
        result = await models.flow_run_notification_policies.delete_flow_run_notification_policy(
            session=session,
            flow_run_notification_policy_id=flow_run_notification_policy_id,
        )
    if not result:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="flow run notification policy not found",
        )

read_flow_run_notification_policies(limit=dependencies.LimitBody(), flow_run_notification_policy_filter=None, offset=Body(0, ge=0), db=Depends(provide_database_interface)) async

Query for flow run notification policies.

Source code in src/prefect/server/api/flow_run_notification_policies.py 
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@router.post("/filter")
async def read_flow_run_notification_policies(
    limit: int = dependencies.LimitBody(),
    flow_run_notification_policy_filter: schemas.filters.FlowRunNotificationPolicyFilter = None,
    offset: int = Body(0, ge=0),
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> List[schemas.core.FlowRunNotificationPolicy]:
    """
    Query for flow run notification policies.
    """
    async with db.session_context() as session:
        return await models.flow_run_notification_policies.read_flow_run_notification_policies(
            session=session,
            flow_run_notification_policy_filter=flow_run_notification_policy_filter,
            offset=offset,
            limit=limit,
        )

read_flow_run_notification_policy(flow_run_notification_policy_id=Path(..., description='The flow run notification policy id', alias='id'), db=Depends(provide_database_interface)) async

Get a flow run notification policy by id.

Source code in src/prefect/server/api/flow_run_notification_policies.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@router.get("/{id}")
async def read_flow_run_notification_policy(
    flow_run_notification_policy_id: UUID = Path(
        ..., description="The flow run notification policy id", alias="id"
    ),
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> schemas.core.FlowRunNotificationPolicy:
    """
    Get a flow run notification policy by id.
    """
    async with db.session_context() as session:
        flow_run_notification_policy = await models.flow_run_notification_policies.read_flow_run_notification_policy(
            session=session,
            flow_run_notification_policy_id=flow_run_notification_policy_id,
        )
    if not flow_run_notification_policy:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="flow run notification policy not found",
        )
    return flow_run_notification_policy

update_flow_run_notification_policy(flow_run_notification_policy, flow_run_notification_policy_id=Path(..., description='The flow run notification policy id', alias='id'), db=Depends(provide_database_interface)) async

Updates an existing flow run notification policy.

Source code in src/prefect/server/api/flow_run_notification_policies.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@router.patch("/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def update_flow_run_notification_policy(
    flow_run_notification_policy: schemas.actions.FlowRunNotificationPolicyUpdate,
    flow_run_notification_policy_id: UUID = Path(
        ..., description="The flow run notification policy id", alias="id"
    ),
    db: PrefectDBInterface = Depends(provide_database_interface),
):
    """
    Updates an existing flow run notification policy.
    """
    async with db.session_context(begin_transaction=True) as session:
        result = await models.flow_run_notification_policies.update_flow_run_notification_policy(
            session=session,
            flow_run_notification_policy_id=flow_run_notification_policy_id,
            flow_run_notification_policy=flow_run_notification_policy,
        )
    if not result:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Flow run notification policy {id} not found",
        )