Zach Schumacher

    Zach Schumacher

    1 year ago
    running into an error I haven’t seen before when registering flows. Can someone point me in the right direction?
    prefect.utilities.exceptions.ClientError: [{'path': ['create_flow_from_compressed_string'], 'message': '1 validation error for FlowSchema\nrun_config\n  value is not a valid dict (type=type_error.dict)', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]
    Kevin Kho

    Kevin Kho

    1 year ago
    Hey @Zach Schumacher, I haven’t seen this either but looks like your RunConfig is not a valid dict? Could you post your RunConfig?
    Zach Schumacher

    Zach Schumacher

    1 year ago
    sure, i subclassed it, maybe that is what it h8s
    class KubernetesRunConfig(KubernetesRun):
        """A KubernetesRun class that gives some warnings for some common prefect config 'gotchas'"""
    
        def __init__(self, *args, **kwargs):
            if "image_pull_policy" not in kwargs:
                # this is the k8s default anyway, but its nice to be explicit so you can see it in the UI
                kwargs.update(image_pull_policy="IfNotPresent")
    
            if "job_template" not in kwargs and "job_template_path" not in kwargs:
                kwargs.update(job_template=_ETL_NODE_TEMPLATE)
    
            if "labels" not in kwargs or not any(label.startswith("k8s-") for label in kwargs["labels"]):
                warnings.warn(
                    "Registering config with no k8s agent label.  "
                    "Your label must match the agent you want to run on or your flow will never start!"
                )
    
            if "image" not in kwargs:
                warnings.warn(
                    "No image passed.  Is this intentional?  "
                    "If you don't pass an image the flow run will use the base prefect image."
                )
            super().__init__(*args, **kwargs)
    run_config=KubernetesRunConfig(labels=[settings.PREFECT_AGENT_LABEL], image=settings.PREFECT_IMAGE),
    Kevin Kho

    Kevin Kho

    1 year ago
    I get a different error when trying to use this RunConfig:
    ValueError: Flow could not be deserialized successfully. Error was: TypeError('not all arguments converted during string formatting')
    . Will look into a bit more later
    I dug through the code and I don’t think you can subclass RunConfig because it looks for the appropriate schema for the RunConfig and uses that to serialize. Code here . I am confirming with the team.
    Zach Schumacher

    Zach Schumacher

    1 year ago
    ah, i could monkey patch that i guess
    this needs to be implemented to support subclassing
    class MyUberSchema(OneOfSchema):
        type_schemas = {"foo": FooSchema, "bar": BarSchema}
    
        def get_obj_type(self, obj):
            if isinstance(obj, Foo):
                return "foo"
            elif isinstance(obj, Bar):
                return "bar"
            else:
                raise Exception("Unknown object type: {}".format(obj.__class__.__name__))
    we need a
    get_obj_type
    func
    Michael Adkins

    Michael Adkins

    1 year ago
    @Zach Schumacher You might be able to accomplish this with a metaclass or by just naming your subclass to collide with the Prefect one
    I'm not sure if we're going to explicitly support subclassing runconfigs yet
    Zach Schumacher

    Zach Schumacher

    1 year ago
    yeah true, thanks for the suggestion. Yeah, I just snagged that from the marshmallow-oneof docs
    but i get why you wouldn’t want to support that