Skip to content

prefect.server.api.block_schemas

Routes for interacting with block schema objects.

delete_block_schema(block_schema_id=Path(..., description='The block schema id', alias='id'), db=Depends(provide_database_interface), api_version=Depends(dependencies.provide_request_api_version)) async

Delete a block schema by id.

Source code in src/prefect/server/api/block_schemas.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_block_schema(
    block_schema_id: UUID = Path(..., description="The block schema id", alias="id"),
    db: PrefectDBInterface = Depends(provide_database_interface),
    api_version=Depends(dependencies.provide_request_api_version),
):
    """
    Delete a block schema by id.
    """
    async with db.session_context(begin_transaction=True) as session:
        block_schema = await models.block_schemas.read_block_schema(
            session=session, block_schema_id=block_schema_id
        )
        if not block_schema:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND, detail="Block schema not found"
            )

        if block_schema.block_type.is_protected:
            raise HTTPException(
                status.HTTP_403_FORBIDDEN,
                detail="Block schemas for protected block types cannot be deleted.",
            )

        await models.block_schemas.delete_block_schema(
            session=session, block_schema_id=block_schema_id
        )

read_block_schema_by_id(block_schema_id=Path(..., description='The block schema id', alias='id'), db=Depends(provide_database_interface)) async

Get a block schema by id.

Source code in src/prefect/server/api/block_schemas.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@router.get("/{id}")
async def read_block_schema_by_id(
    block_schema_id: UUID = Path(..., description="The block schema id", alias="id"),
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> schemas.core.BlockSchema:
    """
    Get a block schema by id.
    """
    async with db.session_context() as session:
        block_schema = await models.block_schemas.read_block_schema(
            session=session, block_schema_id=block_schema_id
        )
    if not block_schema:
        raise HTTPException(status.HTTP_404_NOT_FOUND, detail="Block schema not found")
    return block_schema

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

Read all block schemas, optionally filtered by type

Source code in src/prefect/server/api/block_schemas.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@router.post("/filter")
async def read_block_schemas(
    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.BlockSchema]:
    """
    Read all block schemas, optionally filtered by type
    """
    async with db.session_context() as session:
        result = await models.block_schemas.read_block_schemas(
            session=session,
            block_schema_filter=block_schemas,
            limit=limit,
            offset=offset,
        )
    return result