Failed to retrieve task state with error: ClientError([{'path': ['get_or_create_task_run_info'], 'message': 'Expected type UUID!, found ""; Could not parse UUID: ', 'extensions': {'code': 'INTERNAL_SERVER_ERROR', 'exception': {'message': 'Expected type UUID!, found ""; Could not parse UUID: ', 'locations': [{'line': 2, 'column': 101}], 'path': None}}}])
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/cloud/task_runner.py", line 154, in initialize_run
    task_run_info = self.client.get_task_run_info(
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 1798, in get_task_run_info
    result = self.graphql(mutation)  # type: Any
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 569, in graphql
    raise ClientError(result["errors"])
prefect.exceptions.ClientError: [{'path': ['get_or_create_task_run_info'], 'message': 'Expected type UUID!, found ""; Could not parse UUID: ', 'extensions': {'code': 'INTERNAL_SERVER_ERROR', 'exception': {'message': 'Expected type UUID!, found ""; Could not parse UUID: ', 'locations': [{'line': 2, 'column': 101}], 'path': None}}}]

pg_db = Parameter("LOGARY_PG_DB", default=getenv("LOGARY_PG_DB", default="analytics"))
@task(nout=2)
def fetch_model(dsn_params: DSNParams) -> Tuple[str, ModelDTO]:
    m = _fetch_model(dsn_params)
    return m.id, m

with Flow(
    "run_mmm",
    state_handlers=[print_state_callback],
) as flow:
    # <https://deepnote.com/project/Media-Mix-Model-5xns-00xTG6nRlUK1f9DfA/%2Ftest_preprocessing.ipynb>

    #
    ########### CONFIG #######

    dsn_params = lambda name: build_dsn_params(
        user=pg_user,
        password=pg_password,
        host=pg_host,
        port=pg_port,
        dbname=pg_db,
        sslmode=pg_sslmode,
        application_name=f"mmm/{name}",
    )
    model_id, model = fetch_model(dsn_params("fetch_model"))

prefect run -p

flow.run

prefect run --name "run_mmm" --execute

prefect register --project project_name -p /path/to/flows/that/need/registration

git clone <https://github.com/PrefectHQ/prefect/tree/task-run-info-missing-id>
pip install ./prefect

prefect = { git = "<https://github.com/PrefectHQ/prefect.git>", branch = "task-run-info-missing-id" }

pip install -U git+<https://github.com/PrefectHQ/prefect@task-run-info-missing-id#egg=prefect>

from typing import Optional

from prefect.client.secrets import Secret as _Secret
from prefect.core.task import Task
from prefect.engine.results.secret_result import SecretResult


class LogarySecret(Task):
    """
    Prefect Secrets Task.  This task retrieves the underlying secret through
    the Prefect Secrets API (which has the ability to toggle between local vs. Cloud secrets).

    Args:
        - name (str, optional): The name of the underlying secret
        - **kwargs (Any, optional): additional keyword arguments to pass to the Task constructor

    Raises:
        - ValueError: if a `result` keyword is passed
    """

    secret_name: Optional[str]
    default: Optional[str]

    def __init__(self, name=None, default=None, **kwargs):
        if kwargs.get("result"):
            raise ValueError("Result types for Secrets are not configurable.")
        kwargs["checkpoint"] = False
        self.secret_name = name
        self.default = default
        super().__init__(name=name, **kwargs)
        self.result = SecretResult(secret_task=self)

    def run(self, name: str = None):
        """
        The run method for Secret Tasks.  This method actually retrieves and returns the
        underlying secret value using the `Secret.get()` method.  Note that this method first
        checks context for the secret value, and if not found either raises an error or queries
        Prefect Cloud, depending on whether `config.cloud.use_local_secrets` is `True` or
        `False`.

        Args:
            - name (str, optional): the name of the underlying Secret to retrieve. Defaults
                to the name provided at initialization.

        Returns:
            - Any: the underlying value of the Prefect Secret
        """
        if name is None:
            name = self.secret_name

        if name is None:
            raise ValueError("A secret name must be provided.")

        _s = _Secret(name)
        if not _s.exists() and self.default is not None:
            return self.default

        return _s.get()

build_dsn_params

-1

flow.register