prefect.server.models.block_schemas

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

MissingBlockTypeException

Bases: Exception

Raised when the block type corresponding to a block schema cannot be found

Source code in src/prefect/server/models/block_schemas.py

class MissingBlockTypeException(Exception):
    """Raised when the block type corresponding to a block schema cannot be found"""

create_block_schema(db, session, block_schema, override=False, definitions=None) async

Create a new block schema.

Parameters:

Name	Type	Description	Default
session	AsyncSession	A database session	required
block_schema	Union[BlockSchemaCreate, BlockSchemaCreate]	a block schema object	required
definitions	Optional[Dict]	Definitions of fields from block schema fields attribute. Used when recursively creating nested block schemas	None

Returns:

Name	Type	Description
block_schema		an ORM block schema model

Source code in src/prefect/server/models/block_schemas.py

@db_injector
async def create_block_schema(
    db: PrefectDBInterface,
    session: AsyncSession,
    block_schema: Union[schemas.actions.BlockSchemaCreate, "ClientBlockSchemaCreate"],
    override: bool = False,
    definitions: Optional[Dict] = None,
):
    """
    Create a new block schema.

    Args:
        session: A database session
        block_schema: a block schema object
        definitions: Definitions of fields from block schema fields
            attribute. Used when recursively creating nested block schemas

    Returns:
        block_schema: an ORM block schema model
    """
    from prefect.blocks.core import Block, _get_non_block_reference_definitions

    # We take a shortcut in many unit tests and in block registration to pass client
    # models directly to this function.  We will support this by converting them to
    # the appropriate server model.
    if not isinstance(block_schema, schemas.actions.BlockSchemaCreate):
        block_schema = schemas.actions.BlockSchemaCreate.model_validate(
            block_schema.model_dump(
                mode="json",
                exclude={"id", "created", "updated", "checksum", "block_type"},
            )
        )

    insert_values = block_schema.model_dump_for_orm(
        exclude_unset=False,
        exclude={"block_type", "id", "created", "updated"},
    )

    definitions = definitions or block_schema.fields.get("definitions")
    fields_for_checksum = insert_values["fields"]
    if definitions:
        # Ensure definitions are available if this is a nested schema
        # that is being registered
        fields_for_checksum["definitions"] = definitions
    checksum = Block._calculate_schema_checksum(fields_for_checksum)

    # Check for existing block schema based on calculated checksum
    existing_block_schema = await read_block_schema_by_checksum(
        session=session, checksum=checksum, version=block_schema.version
    )
    # Return existing block schema if it exists. Allows block schema creation to be called multiple
    # times for the same schema without errors.
    if existing_block_schema:
        return existing_block_schema

    insert_values["checksum"] = checksum

    if definitions:
        # Get non block definitions for saving to the DB.
        non_block_definitions = _get_non_block_reference_definitions(
            insert_values["fields"], definitions
        )
        if non_block_definitions:
            insert_values["fields"][
                "definitions"
            ] = _get_non_block_reference_definitions(
                insert_values["fields"], definitions
            )
        else:
            # Prevent storing definitions for blocks. Those are reconstructed on read.
            insert_values["fields"].pop("definitions", None)

    # Prevent saving block schema references in the block_schema table. They have
    # their own table.
    block_schema_references: Dict = insert_values["fields"].pop(
        "block_schema_references", {}
    )

    insert_stmt = db.insert(orm_models.BlockSchema).values(**insert_values)
    if override:
        insert_stmt = insert_stmt.on_conflict_do_update(
            index_elements=db.block_schema_unique_upsert_columns,
            set_=insert_values,
        )
    await session.execute(insert_stmt)

    query = (
        sa.select(orm_models.BlockSchema)
        .where(
            orm_models.BlockSchema.checksum == insert_values["checksum"],
        )
        .order_by(orm_models.BlockSchema.created.desc())
        .limit(1)
        .execution_options(populate_existing=True)
    )

    if block_schema.version is not None:
        query = query.where(orm_models.BlockSchema.version == block_schema.version)

    result = await session.execute(query)
    created_block_schema = copy(result.scalar())

    await _register_nested_block_schemas(
        session=session,
        parent_block_schema_id=created_block_schema.id,
        block_schema_references=block_schema_references,
        base_fields=insert_values["fields"],
        definitions=definitions,
        override=override,
    )

    created_block_schema.fields["block_schema_references"] = block_schema_references
    if definitions is not None:
        created_block_schema.fields["definitions"] = definitions

    return created_block_schema

create_block_schema_reference(db, session, block_schema_reference) async

Retrieves a list of all available block capabilities.

Parameters:

Name	Type	Description	Default
session	AsyncSession	A database session.	required
block_schema_reference	BlockSchemaReference	A block schema reference object.	required

Returns:

Type	Description
	orm_models.BlockSchemaReference: The created BlockSchemaReference

Source code in src/prefect/server/models/block_schemas.py

@db_injector
async def create_block_schema_reference(
    db: PrefectDBInterface,
    session: AsyncSession,
    block_schema_reference: schemas.core.BlockSchemaReference,
):
    """
    Retrieves a list of all available block capabilities.

    Args:
        session: A database session.
        block_schema_reference: A block schema reference object.

    Returns:
        orm_models.BlockSchemaReference: The created BlockSchemaReference
    """
    query_stmt = sa.select(orm_models.BlockSchemaReference).where(
        orm_models.BlockSchemaReference.name == block_schema_reference.name,
        orm_models.BlockSchemaReference.parent_block_schema_id
        == block_schema_reference.parent_block_schema_id,
        orm_models.BlockSchemaReference.reference_block_schema_id
        == block_schema_reference.reference_block_schema_id,
    )

    existing_reference = (await session.execute(query_stmt)).scalar()
    if existing_reference:
        return existing_reference

    insert_stmt = db.insert(orm_models.BlockSchemaReference).values(
        **block_schema_reference.model_dump_for_orm(
            exclude_unset=True, exclude={"created", "updated"}
        )
    )
    await session.execute(insert_stmt)

    result = await session.execute(
        sa.select(orm_models.BlockSchemaReference).where(
            orm_models.BlockSchemaReference.id == block_schema_reference.id
        )
    )
    return result.scalar()

delete_block_schema(session, block_schema_id) async

Delete a block schema by id.

Parameters:

Name	Type	Description	Default
session	Session	A database session	required
block_schema_id	UUID	a block schema id	required

Returns:

Name	Type	Description
bool	bool	whether or not the block schema was deleted

Source code in src/prefect/server/models/block_schemas.py

async def delete_block_schema(session: sa.orm.Session, block_schema_id: UUID) -> bool:
    """
    Delete a block schema by id.

    Args:
        session: A database session
        block_schema_id: a block schema id

    Returns:
        bool: whether or not the block schema was deleted
    """

    result = await session.execute(
        delete(orm_models.BlockSchema).where(
            orm_models.BlockSchema.id == block_schema_id
        )
    )
    return result.rowcount > 0

read_available_block_capabilities(db, session) async

Retrieves a list of all available block capabilities.

Parameters:

Name	Type	Description	Default
session	AsyncSession	A database session.	required

Returns:

Type	Description
List[str]	List[str]: List of all available block capabilities.

Source code in src/prefect/server/models/block_schemas.py

@db_injector
async def read_available_block_capabilities(
    db: PrefectDBInterface,
    session: AsyncSession,
) -> List[str]:
    """
    Retrieves a list of all available block capabilities.

    Args:
        session: A database session.

    Returns:
        List[str]: List of all available block capabilities.
    """
    query = sa.select(
        db.json_arr_agg(db.cast_to_json(orm_models.BlockSchema.capabilities.distinct()))
    )
    capability_combinations = (await session.execute(query)).scalars().first() or list()
    if db.uses_json_strings and isinstance(capability_combinations, str):
        capability_combinations = json.loads(capability_combinations)
    return list({c for capabilities in capability_combinations for c in capabilities})

read_block_schema(session, block_schema_id) async

Reads a block schema by id. Will reconstruct the block schema's fields attribute to include block schema references.

Parameters:

Name	Type	Description	Default
session	AsyncSession	A database session	required
block_schema_id	UUID	a block_schema id	required

Returns:

Type	Description
	orm_models..BlockSchema: the block_schema

Source code in src/prefect/server/models/block_schemas.py

async def read_block_schema(
    session: AsyncSession,
    block_schema_id: UUID,
):
    """
    Reads a block schema by id. Will reconstruct the block schema's fields attribute
    to include block schema references.

    Args:
        session: A database session
        block_schema_id: a block_schema id

    Returns:
        orm_models..BlockSchema: the block_schema
    """

    # Construction of a recursive query which returns the specified block schema
    # along with and nested block schemas coupled with the ID of their parent schema
    # the key that they reside under.
    block_schema_references_query = (
        sa.select(orm_models.BlockSchemaReference)
        .select_from(orm_models.BlockSchemaReference)
        .filter_by(parent_block_schema_id=block_schema_id)
        .cte("block_schema_references", recursive=True)
    )
    block_schema_references_join = (
        sa.select(orm_models.BlockSchemaReference)
        .select_from(orm_models.BlockSchemaReference)
        .join(
            block_schema_references_query,
            orm_models.BlockSchemaReference.parent_block_schema_id
            == block_schema_references_query.c.reference_block_schema_id,
        )
    )
    recursive_block_schema_references_cte = block_schema_references_query.union_all(
        block_schema_references_join
    )
    nested_block_schemas_query = (
        sa.select(
            orm_models.BlockSchema,
            recursive_block_schema_references_cte.c.name,
            recursive_block_schema_references_cte.c.parent_block_schema_id,
        )
        .select_from(orm_models.BlockSchema)
        .join(
            recursive_block_schema_references_cte,
            orm_models.BlockSchema.id
            == recursive_block_schema_references_cte.c.reference_block_schema_id,
            isouter=True,
        )
        .filter(
            sa.or_(
                orm_models.BlockSchema.id == block_schema_id,
                recursive_block_schema_references_cte.c.parent_block_schema_id.is_not(
                    None
                ),
            )
        )
    )
    result = await session.execute(nested_block_schemas_query)

    return _construct_full_block_schema(result.all())

read_block_schema_by_checksum(session, checksum, version=None) async

Reads a block_schema by checksum. Will reconstruct the block schema's fields attribute to include block schema references.

Parameters:

Name	Type	Description	Default
session	AsyncSession	A database session	required
checksum	str	a block_schema checksum	required
version	Optional[str]	A block_schema version	None

Returns:

Type	Description
Optional[BlockSchema]	orm_models.BlockSchema: the block_schema

Source code in src/prefect/server/models/block_schemas.py

async def read_block_schema_by_checksum(
    session: AsyncSession,
    checksum: str,
    version: Optional[str] = None,
) -> Optional[BlockSchema]:
    """
    Reads a block_schema by checksum. Will reconstruct the block schema's fields
    attribute to include block schema references.

    Args:
        session: A database session
        checksum: a block_schema checksum
        version: A block_schema version

    Returns:
        orm_models.BlockSchema: the block_schema
    """
    # Construction of a recursive query which returns the specified block schema
    # along with and nested block schemas coupled with the ID of their parent schema
    # the key that they reside under.

    # The same checksum with different versions can occur in the DB. Return only the
    # most recently created one.
    root_block_schema_query = (
        sa.select(orm_models.BlockSchema)
        .filter_by(checksum=checksum)
        .order_by(orm_models.BlockSchema.created.desc())
        .limit(1)
    )

    if version is not None:
        root_block_schema_query = root_block_schema_query.filter_by(version=version)

    root_block_schema_cte = root_block_schema_query.cte("root_block_schema")

    block_schema_references_query = (
        sa.select(orm_models.BlockSchemaReference)
        .select_from(orm_models.BlockSchemaReference)
        .filter_by(parent_block_schema_id=root_block_schema_cte.c.id)
        .cte("block_schema_references", recursive=True)
    )
    block_schema_references_join = (
        sa.select(orm_models.BlockSchemaReference)
        .select_from(orm_models.BlockSchemaReference)
        .join(
            block_schema_references_query,
            orm_models.BlockSchemaReference.parent_block_schema_id
            == block_schema_references_query.c.reference_block_schema_id,
        )
    )
    recursive_block_schema_references_cte = block_schema_references_query.union_all(
        block_schema_references_join
    )
    nested_block_schemas_query = (
        sa.select(
            orm_models.BlockSchema,
            recursive_block_schema_references_cte.c.name,
            recursive_block_schema_references_cte.c.parent_block_schema_id,
        )
        .select_from(orm_models.BlockSchema)
        .join(
            recursive_block_schema_references_cte,
            orm_models.BlockSchema.id
            == recursive_block_schema_references_cte.c.reference_block_schema_id,
            isouter=True,
        )
        .filter(
            sa.or_(
                orm_models.BlockSchema.id == root_block_schema_cte.c.id,
                recursive_block_schema_references_cte.c.parent_block_schema_id.is_not(
                    None
                ),
            )
        )
    )
    result = await session.execute(nested_block_schemas_query)
    return _construct_full_block_schema(result.all())

read_block_schemas(session, block_schema_filter=None, limit=None, offset=None) async

Reads block schemas, optionally filtered by type or name.

Parameters:

Name	Type	Description	Default
session	AsyncSession	A database session	required
block_schema_filter	Optional[BlockSchemaFilter]	a block schema filter object	None
limit	int	query limit	None
offset	int	query offset	None

Returns:

Type	Description
	List[orm_models.BlockSchema]: the block_schemas

Source code in src/prefect/server/models/block_schemas.py

async def read_block_schemas(
    session: AsyncSession,
    block_schema_filter: Optional[schemas.filters.BlockSchemaFilter] = None,
    limit: Optional[int] = None,
    offset: Optional[int] = None,
):
    """
    Reads block schemas, optionally filtered by type or name.

    Args:
        session: A database session
        block_schema_filter: a block schema filter object
        limit (int): query limit
        offset (int): query offset

    Returns:
        List[orm_models.BlockSchema]: the block_schemas
    """
    # schemas are ordered by `created DESC` to get the most recently created
    # ones first (and to facilitate getting the newest one with `limit=1`).
    filtered_block_schemas_query = select(orm_models.BlockSchema.id).order_by(
        orm_models.BlockSchema.created.desc()
    )

    if block_schema_filter:
        filtered_block_schemas_query = filtered_block_schemas_query.where(
            block_schema_filter.as_sql_filter()
        )

    if offset is not None:
        filtered_block_schemas_query = filtered_block_schemas_query.offset(offset)
    if limit is not None:
        filtered_block_schemas_query = filtered_block_schemas_query.limit(limit)

    filtered_block_schema_ids = (
        (await session.execute(filtered_block_schemas_query)).scalars().unique().all()
    )

    block_schema_references_query = (
        sa.select(orm_models.BlockSchemaReference)
        .select_from(orm_models.BlockSchemaReference)
        .filter(
            orm_models.BlockSchemaReference.parent_block_schema_id.in_(
                filtered_block_schemas_query
            )
        )
        .cte("block_schema_references", recursive=True)
    )
    block_schema_references_join = (
        sa.select(orm_models.BlockSchemaReference)
        .select_from(orm_models.BlockSchemaReference)
        .join(
            block_schema_references_query,
            orm_models.BlockSchemaReference.parent_block_schema_id
            == block_schema_references_query.c.reference_block_schema_id,
        )
    )
    recursive_block_schema_references_cte = block_schema_references_query.union_all(
        block_schema_references_join
    )

    nested_block_schemas_query = (
        sa.select(
            orm_models.BlockSchema,
            recursive_block_schema_references_cte.c.name,
            recursive_block_schema_references_cte.c.parent_block_schema_id,
        )
        .select_from(orm_models.BlockSchema)
        # in order to reconstruct nested block schemas efficiently, we need to visit them
        # in the order they were created (so that we guarantee that nested/referenced schemas)
        # have already been seen. Therefore this second query sorts by created ASC
        .order_by(orm_models.BlockSchema.created.asc())
        .join(
            recursive_block_schema_references_cte,
            orm_models.BlockSchema.id
            == recursive_block_schema_references_cte.c.reference_block_schema_id,
            isouter=True,
        )
        .filter(
            sa.or_(
                orm_models.BlockSchema.id.in_(filtered_block_schemas_query),
                recursive_block_schema_references_cte.c.parent_block_schema_id.is_not(
                    None
                ),
            )
        )
    )

    block_schemas_with_references = (
        (await session.execute(nested_block_schemas_query)).unique().all()
    )
    fully_constructed_block_schemas = []
    visited_block_schema_ids = []
    for root_block_schema, _, _ in block_schemas_with_references:
        if (
            root_block_schema.id in filtered_block_schema_ids
            and root_block_schema.id not in visited_block_schema_ids
        ):
            fully_constructed_block_schemas.append(
                _construct_full_block_schema(
                    block_schemas_with_references=block_schemas_with_references,
                    root_block_schema=root_block_schema,
                )
            )
            visited_block_schema_ids.append(root_block_schema.id)

    # because we reconstructed schemas ordered by created ASC, we
    # reverse the final output to restore created DESC
    return list(reversed(fully_constructed_block_schemas))