Gather new data, process if new data or wait if no new data

[srtenvex]

Hi,
Just discovering the package. A huge fan already.

I know that new data from an API will come between 13.30 and 15.00 daily , and is there a way I can set the task to fail and then retry again until new data is available? And then if there is new data then set it to success and process the data?

Or is there a better way to implement this in the rocketry framework?

Like for instance:

@app.task(daily.at("13:30"))
async def gather_day_prices():
    # gather some data

    if not new_data:
         # Set the task to fail 
    if new_data:
         # Set the task to be succesfull
         
@app.task(after_fail(gather_day_prices))
def do_after_fail():
       # sleep for a bit and then try again
       time.sleep(120)
       gather_day_prices()

@app.task(after_success(gather_day_prices))
def do_after_success():
      # process data
       

[Miksus]

Hi!

Thanks! It's always nice to hear people finding it useful and liking my ideas :)

I have been prototyping some sort of an event model (which is just extending the current functionalities) but this is still quite on the drawing board.

I would probably try to do this via (custom) conditions. There is also a task status inaction which could turn out to be useful but that's not much documented as I'm not sure if it will be in the package in the long run (it can be useful but causes a [very] small burden for most uses).

There is also the rocketry.conds.retry which could be useful for you but that does not support periods to wait at the moment (though maybe it could in the future).

So maybe something like this:

from rocketry.args import Return

@app.cond()
def has_new_data(data=Return("gather_day_prices", default=None)):
    has_data = data is not None
    is_processed = ... # TODO
    return has_data  and not is_processed 

@app.task(daily.at("13:30"))
async def gather_day_prices():
    # gather some data
    # We just simulate that 50% of the time there is new data

    import random
    print("Gathering data")
    data = "some data" if random.random() > 0.5 else None

    return data

@app.task(has_new_data)
def process_data(data=Return(gather_day_prices)):
    # process data
    print("Processing data")

I can try to revisit this (and the event mechanics are still on my table) but need to go climbing now. I hope this gives you some ideas to go forward. It still needs working on how to do the retrying.

[srtenvex]

Thank u. I ended up implementing a custom condition per your comment. It works rather well now in combination with my database and FastAPI appilcation setup. And sounds interesting with the event model if you ever need someone to do some testing on your prototype please not hesitate to reach out :)