prefect.workers.block

BlockWorkerJobConfiguration

Bases: BaseModel

Source code in prefect/workers/block.py

class BlockWorkerJobConfiguration(BaseModel):
    block: Block = Field(
        default=..., description="The infrastructure block to use for job creation."
    )

    @validator("block")
    def _validate_infrastructure_block(cls, v):
        return validate_block_is_infrastructure(v)

    _related_objects: Dict[str, Any] = PrivateAttr(default_factory=dict)

    @property
    def is_using_a_runner(self):
        return (
            self.block.command is not None
            and "prefect flow-run execute" in shlex.join(self.block.command)
        )

    @staticmethod
    def _get_base_config_defaults(variables: dict) -> dict:
        """Get default values from base config for all variables that have them."""
        defaults = dict()
        for variable_name, attrs in variables.items():
            if "default" in attrs:
                defaults[variable_name] = attrs["default"]

        return defaults

    @classmethod
    @inject_client
    async def from_template_and_values(
        cls, base_job_template: dict, values: dict, client: "PrefectClient" = None
    ):
        """Creates a valid worker configuration object from the provided base
        configuration and overrides.

        Important: this method expects that the base_job_template was already
        validated server-side.
        """
        job_config: Dict[str, Any] = base_job_template["job_configuration"]
        variables_schema = base_job_template["variables"]
        variables = cls._get_base_config_defaults(
            variables_schema.get("properties", {})
        )
        variables.update(values)

        populated_configuration = apply_values(template=job_config, values=variables)

        block_document_id = get_from_dict(
            populated_configuration, "block.$ref.block_document_id"
        )
        if not block_document_id:
            raise ValueError(
                "Base job template is invalid for this worker type because it does not"
                " contain a block_document_id after variable resolution."
            )

        block_document = await client.read_block_document(
            block_document_id=block_document_id
        )
        infrastructure_block = Block._from_block_document(block_document)

        populated_configuration["block"] = infrastructure_block

        return cls(**populated_configuration)

    @classmethod
    def json_template(cls) -> dict:
        """Returns a dict with job configuration as keys and the corresponding templates as values

        Defaults to using the job configuration parameter name as the template variable name.

        e.g.
        {
            key1: '{{ key1 }}',     # default variable template
            key2: '{{ template2 }}', # `template2` specifically provide as template
        }
        """
        configuration = {}
        properties = cls.schema()["properties"]
        for k, v in properties.items():
            if v.get("template"):
                template = v["template"]
            else:
                template = "{{ " + k + " }}"
            configuration[k] = template

        return configuration

    def _related_resources(self) -> List[RelatedResource]:
        tags = set()
        related = []

        for kind, obj in self._related_objects.items():
            if obj is None:
                continue
            if hasattr(obj, "tags"):
                tags.update(obj.tags)
            related.append(object_as_related_resource(kind=kind, role=kind, object=obj))

        return related + tags_as_related_resources(tags)

    def prepare_for_flow_run(
        self,
        flow_run: "FlowRun",
        deployment: Optional["DeploymentResponse"] = None,
        flow: Optional["Flow"] = None,
    ):
        self.block = self.block.prepare_for_flow_run(
            flow_run=flow_run, deployment=deployment, flow=flow
        )

from_template_and_values async classmethod

Creates a valid worker configuration object from the provided base configuration and overrides.

Important: this method expects that the base_job_template was already validated server-side.

Source code in prefect/workers/block.py

@classmethod
@inject_client
async def from_template_and_values(
    cls, base_job_template: dict, values: dict, client: "PrefectClient" = None
):
    """Creates a valid worker configuration object from the provided base
    configuration and overrides.

    Important: this method expects that the base_job_template was already
    validated server-side.
    """
    job_config: Dict[str, Any] = base_job_template["job_configuration"]
    variables_schema = base_job_template["variables"]
    variables = cls._get_base_config_defaults(
        variables_schema.get("properties", {})
    )
    variables.update(values)

    populated_configuration = apply_values(template=job_config, values=variables)

    block_document_id = get_from_dict(
        populated_configuration, "block.$ref.block_document_id"
    )
    if not block_document_id:
        raise ValueError(
            "Base job template is invalid for this worker type because it does not"
            " contain a block_document_id after variable resolution."
        )

    block_document = await client.read_block_document(
        block_document_id=block_document_id
    )
    infrastructure_block = Block._from_block_document(block_document)

    populated_configuration["block"] = infrastructure_block

    return cls(**populated_configuration)

json_template classmethod

Returns a dict with job configuration as keys and the corresponding templates as values

Defaults to using the job configuration parameter name as the template variable name.

e.g. { key1: '{{ key1 }}', # default variable template key2: '{{ template2 }}', # template2 specifically provide as template }

Source code in prefect/workers/block.py

@classmethod
def json_template(cls) -> dict:
    """Returns a dict with job configuration as keys and the corresponding templates as values

    Defaults to using the job configuration parameter name as the template variable name.

    e.g.
    {
        key1: '{{ key1 }}',     # default variable template
        key2: '{{ template2 }}', # `template2` specifically provide as template
    }
    """
    configuration = {}
    properties = cls.schema()["properties"]
    for k, v in properties.items():
        if v.get("template"):
            template = v["template"]
        else:
            template = "{{ " + k + " }}"
        configuration[k] = template

    return configuration

BlockWorkerResult

Bases: BaseWorkerResult

Result of a block worker job

Source code in prefect/workers/block.py

class BlockWorkerResult(BaseWorkerResult):
    """Result of a block worker job"""