Hi folks, I'm working locally with Prefect core, stringing together a bunch of functions that hit twitter's API- with a screen name, get the user object; get the user's followees; get tweets of the user's followees; get urls mentioned in tweets, etc. Each function is task decorated, and since these are fan out operations each step in the flow uses map on the results of the previous step. Since twitter has strict rate limits I want to cache outputs of each task during testing, so am using the cache_for option in the task decorator, with an hour as the duration...and this does not seem to be working. Every flow run hits the twitter api rather than pulling results from a cache, and I see "cache is now invalid" in the logs: 2019-08-22 19:31:25,037] INFO - prefect.TaskRunner | Task 'auth_name_to_id': Starting task run... [2019-08-22 19:31:25,038] INFO - prefect.TaskRunner | Task 'auth_name_to_id[1]': Starting task run... [2019-08-22 19:31:25,038] WARNING - prefect.TaskRunner | Task 'auth_name_to_id[1]': can't use cache because it is now invalid [2019-08-22 19:31:25,200] INFO - prefect.TaskRunner | Task 'auth_name_to_id[1]': finished task run for task with final state: 'Cached' [2019-08-22 19:31:25,200] INFO - prefect.TaskRunner | Task 'auth_name_to_id[0]': Starting task run... [2019-08-22 19:31:25,200] WARNING - prefect.TaskRunner | Task 'auth_name_to_id[0]': can't use cache because it is now invalid [2019-08-22 19:31:25,358] INFO - prefect.TaskRunner | Task 'auth_name_to_id[0]': finished task run for task with final state: 'Cached' [2019-08-22 19:31:25,359] INFO - prefect.TaskRunner | Task 'auth_name_to_id': finished task run for task with final state: 'Mapped' [2019-08-22 19:31:25,359] INFO - prefect.TaskRunner | Task 'auth_to_friend': Starting task run... [2019-08-22 19:31:25,360] INFO - prefect.TaskRunner | Task 'auth_to_friend[0]': Starting task run... [2019-08-22 19:31:25,360] WARNING - prefect.TaskRunner | Task 'auth_to_friend[0]': can't use cache because it is now invalid [2019-08-22 19:31:25,605] INFO - prefect.TaskRunner | Unexpected error: TwitterError([{'message': 'Rate limit exceeded', 'code': 88}]) [2019-08-22 19:31:25,605] INFO - prefect.TaskRunner | Task 'auth_to_friend[1]': finished task run for task with final state: 'Retrying' .... Should something like this just work? Without looking too closely at the code, the thought that comes to mind is that the functions themselves take dicts, pulling out particular keys, and return lists of dicts. I don't work a lot in python and don't know how it handles equality tests of dicts- is the use of dicts in the task calls potentially screwing up caching?

@Jonah Benton are you running on 0.6.0? There was a bug that caused the log statement about invalid caches to print even when the cache was valid

Looks like 0.6.1: pip3 list --user | grep prefect prefect 0.6.1

Ah, I apologize then

I want to cache, and it doesn't seem to be doing so, as I am running into rate limits

That’s certainly odd. All of your flow runs are in the same python process? Caches in core are shared across the process.

Is caching memory based? I am running flows by running a script with flow.run() at the end. So new process per flow execution.

Ok, that’s the likely culprit in this case. The caching mechanism as implemented in Core is meant principally for testing flows that run on a schedule with

flow.run()

in a single, long-running process, so they are cached in memory. Caching is really meant to be used with a distributed, stateful backend so that cached results can be retrieved as needed via API no matter where a deployed flow is running (which is how you’d use it in Prefect Cloud).

Ok, got it, thank you. I am not at the place where I could benefit from access to cloud but will definitely ask if we get there. Is the cache pluggable, can I put in a quick file system based get/put thingee?

It’s pluggable but a little bit intricately tied to some other (also pluggable) things like

ResultHandlers

. There’s already a filesystem resulthandler so I suspect that @Chris White will tell us this is actually a lot easier than it seems

Cool, thank you

👋 I would actually recommend using a

state_handler

on your tasks which require caching; if the task is in a

Pending

-like state, you can extract the stored

Cached

state from your local filesystem and return it from the handler, allowing Prefect Core to determine if the cache is still valid or not

Thank you, makes sense!