Skip to content

prefect.server.api.block_types

create_block_type(block_type, db=Depends(provide_database_interface)) async

Create a new block type

Source code in src/prefect/server/api/block_types.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@router.post("/", status_code=status.HTTP_201_CREATED)
async def create_block_type(
    block_type: schemas.actions.BlockTypeCreate,
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> schemas.core.BlockType:
    """
    Create a new block type
    """
    # API-created blocks cannot start with the word "Prefect"
    # as it is reserved for system use
    if block_type.name.lower().startswith("prefect"):
        raise HTTPException(
            status.HTTP_403_FORBIDDEN,
            detail="Block type names beginning with 'Prefect' are reserved.",
        )
    try:
        async with db.session_context(begin_transaction=True) as session:
            created_block_type = await models.block_types.create_block_type(
                session, block_type=block_type
            )
    except sa.exc.IntegrityError:
        raise HTTPException(
            status.HTTP_409_CONFLICT,
            detail=f'Block type with name "{block_type.name}" already exists',
        )
    return created_block_type

read_block_type_by_id(block_type_id=Path(..., description='The block type ID', alias='id'), db=Depends(provide_database_interface)) async

Get a block type by ID.

Source code in src/prefect/server/api/block_types.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@router.get("/{id}")
async def read_block_type_by_id(
    block_type_id: UUID = Path(..., description="The block type ID", alias="id"),
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> schemas.core.BlockType:
    """
    Get a block type by ID.
    """
    async with db.session_context() as session:
        block_type = await models.block_types.read_block_type(
            session=session, block_type_id=block_type_id
        )
    if not block_type:
        raise HTTPException(status.HTTP_404_NOT_FOUND, detail="Block type not found")
    return block_type

read_block_type_by_slug(block_type_slug=Path(..., description='The block type name', alias='slug'), db=Depends(provide_database_interface)) async

Get a block type by name.

Source code in src/prefect/server/api/block_types.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
@router.get("/slug/{slug}")
async def read_block_type_by_slug(
    block_type_slug: str = Path(..., description="The block type name", alias="slug"),
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> schemas.core.BlockType:
    """
    Get a block type by name.
    """
    async with db.session_context() as session:
        block_type = await models.block_types.read_block_type_by_slug(
            session=session, block_type_slug=block_type_slug
        )
    if not block_type:
        raise HTTPException(status.HTTP_404_NOT_FOUND, detail="Block type not found")
    return block_type

read_block_types(block_types=None, block_schemas=None, limit=dependencies.LimitBody(), offset=Body(0, ge=0), db=Depends(provide_database_interface)) async

Gets all block types. Optionally limit return with limit and offset.

Source code in src/prefect/server/api/block_types.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
@router.post("/filter")
async def read_block_types(
    block_types: Optional[schemas.filters.BlockTypeFilter] = None,
    block_schemas: Optional[schemas.filters.BlockSchemaFilter] = None,
    limit: int = dependencies.LimitBody(),
    offset: int = Body(0, ge=0),
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> List[schemas.core.BlockType]:
    """
    Gets all block types. Optionally limit return with limit and offset.
    """
    async with db.session_context() as session:
        return await models.block_types.read_block_types(
            session=session,
            limit=limit,
            offset=offset,
            block_type_filter=block_types,
            block_schema_filter=block_schemas,
        )

update_block_type(block_type, block_type_id=Path(..., description='The block type ID', alias='id'), db=Depends(provide_database_interface)) async

Update a block type.

Source code in src/prefect/server/api/block_types.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
@router.patch("/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def update_block_type(
    block_type: schemas.actions.BlockTypeUpdate,
    block_type_id: UUID = Path(..., description="The block type ID", alias="id"),
    db: PrefectDBInterface = Depends(provide_database_interface),
):
    """
    Update a block type.
    """
    async with db.session_context(begin_transaction=True) as session:
        db_block_type = await models.block_types.read_block_type(
            session=session, block_type_id=block_type_id
        )
        if db_block_type is None:
            raise HTTPException(
                status.HTTP_404_NOT_FOUND, detail="Block type not found"
            )

        # Only update the block type if there is any meaningful changes.
        # This avoids deadlocks when creating multiple blocks of the same type.
        # This check happens client side, but we do it server side as well
        # to accommodate older clients.
        if _should_update_block_type(
            block_type,
            schemas.core.BlockType.model_validate(db_block_type, from_attributes=True),
        ):
            await models.block_types.update_block_type(
                session=session, block_type=block_type, block_type_id=block_type_id
            )