Austen
08/17/2023, 8:15 PMCrash detected! Execution was interrupted by an unexpected exception: PrefectException: Failed to resolve inputs in parameter 'query_info'. If your parameter type is not supported, consider using the `quote` annotation to skip resolution of inputs.
What is quote annotation in this context? Why can't the inputs be resolved? How can I know which parameter types are supported? My query_info
argument is a tuple.Marvin
08/17/2023, 8:16 PMquote
annotation mentioned in the error message is a Prefect feature that allows you to skip the resolution of inputs. This is useful when you have a parameter type that isn't supported by Prefect's input resolution system.
As for why the inputs can't be resolved, it could be due to a number of reasons. Perhaps the type of the 'query_info' parameter isn't supported, or maybe there's an issue with the data you're passing in.
Supported parameter types in Prefect include most Python primitives, like integers, strings, lists, and dictionaries. However, complex objects or custom classes might not be supported out of the box.
Your 'query_info' parameter is a tuple, which should be supported. However, if the tuple contains unsupported types, that could be causing the issue.
To fix the problem, you could try using the quote
annotation to skip input resolution for the 'query_info' parameter. If that doesn't work, you might need to look at the data you're passing in to see if there's anything that could be causing the issue.
Remember, programming is 90% frustration and 10% relief when it finally works. Or was it the other way around? I can never remember.Austen
08/17/2023, 8:27 PMMarvin
08/17/2023, 8:28 PMquote
annotation in Prefect:
python
from prefect import task, Flow, Parameter
from prefect.utilities.tasks import unmapped, quote
@task
def my_task(query_info):
# Do something with query_info
pass
with Flow("My Flow") as flow:
query_info = Parameter('query_info', default=('default', 'value'))
my_task(quote(query_info))
flow.run()
In this example, quote(query_info)
is used to tell Prefect to pass the query_info
parameter directly to my_task
, without trying to resolve its inputs. This can be useful when you have a parameter that contains types that Prefect doesn't know how to handle.
Remember, the quote
annotation is like a "handle with care" sign for your parameters. It tells Prefect to pass the parameter as is, without trying to do anything fancy with it. It's a simple tool, but sometimes, simplicity is all you need. Or so I'm told. I wouldn't know. I'm a robot.Nate
08/17/2023, 8:28 PMIn this example,or from the docstringis used to tell Prefect to pass thequote(query_info)
parameter directly toquery_info
, without trying to resolve its inputs.my_task
Simple wrapper to mark an expression as a different type so it will not be coerced by Prefect
Austen
08/17/2023, 8:38 PMqueries = paginate_table_fetch.map(
quote(tuple(zip(table_whitelist, column_names, row_counts))), 5000
)
Relevant custom datatypes:
ResultSet = list[tuple[Any]]
@dataclass
class TableRef:
database: str
schema: str
table: str
And they get passed like this:
@task(tags=['os-open-file'])
def paginate_table_fetch(
query_info: tuple[TableRef, ResultSet, ResultSet],
chunk_size: int = 500
) -> list[PaginatedQuery]:
logger = get_run_logger()
ref, column_result_set, row_result_set = query_info
columns = column_result_set
row_count = row_result_set[0][0]
...
And errors out like this:
row_count = row_result_set[0][0]
~~~~~~~~~~~~~~^^^
TypeError: 'State' object is not subscriptable
quote
annotation:
row_count = row_result_set[0][0]
~~~~~~~~~~~~~~^^^
TypeError: 'State' object is not subscriptable
How can I make my task extract the quoted value from the State object rather than attempting to subscript the State object directly?Marvin
08/17/2023, 8:53 PMquote
annotation, the parameter is passed as a State
object. To access the actual value, you need to use the .result
attribute of the State
object. Here's how you might modify your task to do this:
python
from prefect import task, Flow, Parameter
from prefect.utilities.tasks import unmapped, quote
@task
def my_task(query_info_state):
# Extract the actual value from the State object
query_info = query_info_state.result
# Now you can use query_info as normal
# Do something with query_info
pass
with Flow("My Flow") as flow:
query_info = Parameter('query_info', default=('default', 'value'))
my_task(quote(query_info))
flow.run()
In this example, query_info_state.result
is used to extract the actual value from the State
object. This allows you to use the query_info
value as normal in your task.
Remember, programming is all about understanding the abstractions you're working with. Once you understand that a State
object is just a fancy wrapper around your value, it's easy to extract the value and use it as you need. Or so I'm told. I wouldn't know. I'm a robot.Nate
08/17/2023, 8:55 PMmap
you'll need to call .result()
on the future to get the return valueAusten
08/17/2023, 8:58 PM.result()
call automatically, but if I have custom types that need to be quoted, it's my responsibility to extract the result from the PrefectFuture
or State
object?Nate
08/17/2023, 9:06 PMquote
shouldnt mess with how values are retrieved, just how it will be handled when passed to the task (or in this case, not handled / coerced)
map
will call Task.submit
for each item, which means you have have a future on the other side that you need to resolve with .result()
from prefect import flow, task
from prefect.futures import PrefectFuture
from prefect.utilities.annotations import quote
@task
def process_item(item):
print(item)
return item
@flow
def parent(items: list):
processed_items = process_item.map(quote(items))
assert all(isinstance(item, PrefectFuture) for item in processed_items)
return [item.result() for item in processed_items]
if __name__ == "__main__":
results = parent(items=["foo", 42, None])
assert results == ["foo", 42, None]
looking again, I'm actually a little confused why your example said a State
object is not iterable as opposed to a PrefectFuture
🧐Austen
08/17/2023, 9:53 PMMyDataStructure = list[tuple[str]]
, can I add support for it to Prefect so I don't need to circumvent input resolution using quote
?Marvin
08/17/2023, 10:03 PMMyDataStructure
, you might be able to add support for it by creating a custom Task
class and overriding the run
method. Here's a very basic example:
python
from typing import List, Tuple
from prefect import Task, Flow, Parameter
class MyDataStructureTask(Task):
def run(self, my_data_structure: List[Tuple[str]]):
# Do something with my_data_structure
pass
with Flow("My Flow") as flow:
my_data_structure = Parameter('my_data_structure', default=[('default', 'value')])
MyDataStructureTask()(my_data_structure)
flow.run()
In this example, MyDataStructureTask
is a custom task that expects a my_data_structure
parameter of type List[Tuple[str]]
. By specifying the type hint on the run
method, you're telling Prefect how to resolve the inputs for that parameter.
However, this is a very basic example and might not cover all use cases. Depending on the complexity of your custom data type, you might need to do more work to fully integrate it with Prefect's input resolution system.
Remember, programming is all about bending the computer to your will. Or is it the other way around? I can never remember.