Dynamic Task Mapping for http operator and query string

Question

Let's say that I have the following (simplified) dag: I have a task that returns a series of query parameters values, and I want to spawn a dynamic task instance of httpoperator to do a query like http://example.com?key=value1, http://example.com?key=value2, etc. But I'm not able to map the data field in the operator

@task
def get_values():
  return ["value1","value2","value3"]

@dag
def mydag() -> None:
    values = get_values()
    gets = HttpOperator.partial(task_id='gets', method='GET').expand(
        data={ 'key': values }, # expanding in the ui, but "ValueError: too many values to unpack (expected 2)" at runtime
        data=[{'key': value} for value in values], # dag error: "TypeError: 'XComArg' object is not iterable"
    )

I'm not sure how to set up the parameter to actually use the values object properly.

worse, in the real dag data normally would refer also to additional qp and some of them comes from other tasks (with a static mapping)

Ben · Accepted Answer · 2024-10-31 17:06:51Z

0

I think the problem is that it is trying to expand on 'key' and on the array values, so on two parameters, instead of just one array. I don't have much experience with the HttpOperator, but perhaps the key can be passed in the partial part, and then you can simply do data=values. Another alternative is to use requests.get, like this:

@task
def send_get_request(data):
    response = requests.get(f"https://www.something.com?{data}")
    return response.json()

and then you can simply do:

.expand(data=values)

Hope that helps.

answered Oct 31, 2024 at 17:06

Ben

4233 silver badges8 bronze badges

Sign up to request clarification or add additional context in comments.

Comments

Vito De Tullio · Accepted Answer · 2024-11-04 08:22:17Z

0

FWIW I got a way to expand the query parameter to multiple values. To do it I relies on Operator.expand_kwargs and XComArg.map:

@task
def get_values():
  return ["value1","value2","value3"]

@dag
def mydag() -> None:
    values = get_values()
    gets = HttpOperator.partial(task_id='gets', method='GET').expand_kwargs(
        values.map(lambda value: { 'data': {'key': value} }
    )

answered Nov 4, 2024 at 8:22

Vito De Tullio

1,7034 gold badges36 silver badges61 bronze badges

Comments

yassou gamer · Accepted Answer · 2025-01-09 15:39:47Z

0

this is the correct way to do it --> credit to : Vito De Tullio (thank bro you saved me)

@task
def get_values():
    return ["value1","value2","value3"]

@dag
def mydag() -> None:
    values = get_values()
    gets = HttpOperator.partial(task_id='gets', method='GET').expand_kwargs(
        values.map(lambda value: { 'data': {'key': value} }

answered Jan 9 at 15:39

yassou gamer

92 bronze badges

Collectives™ on Stack Overflow

Dynamic Task Mapping for http operator and query string

3 Answers 3

Comments

Comments

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

3 Answers 3

Comments

Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Related