Hello 🙂 I believe I ran into a threading problem yesterday. I am converting some data extraction pipelines to prefect 2 flows, which works fine for most of them. I just add the decorators and some logging. But one of the data pipelines uses sqlite for intermediate storage, and this is the error message I receive:

sqlite3.ProgrammingError: SQLite objects created in a thread can only be used in that same thread. The object was created in thread id 140193261803264 and this is thread id 140192735962880.

I can also paste the part of the code where sqlite is called. I am also trying to guess the reason behind the error. There is an sqlite object created in a function. Within the function I define another function which uses this object. But both python functions are not prefect flows or tasks. They live within a bigger prefect flow. So here comes my first question: Does prefect 2 create a different thread for each nested function inside a flow or a task. Otherwise I cannot explain why the 2 parts of the code would run in different threads.

Which version of Prefect are you in?

let me check

Can you upgrade to the latest version of Prefect and see if this error still persists?

If the error still happens on 2.3.1, are you able to post any code? Or if not, even example code that shows how you are using the function and nested function would be useful. I have an idea of why this is happening but seeing code would make it easier to confirm. 🙂

Yes I will upgrade and I just made a photo:

does the pipeline execution happen in a Prefect task?

yes it does

I think that's why you are seeing this. Prefect 2's default task runner is

ConcurrentTaskRunner

, which uses a thread pool to run tasks. Your nested function creates a closure around the

sqlite3.Row

created on the main thread, so you see this error when you execute the function later on inside a task

ah I see. yes the closure was somehow necessary because of the lambda I believe. but another option

SequentialTaskRunner

also uses threads? but in sequence?

A couple of things to try: • Call the function that returns this nested function from inside the task that executes the pipeline. Even if this isn't how you want to structure your code in your final flow, it would help narrow down the cause. • Try running the flow using

SeqentialTaskRunner

and see if the problem goes away. Again, it might not be the solution you want to use for your production flow but would provide useful information.

Thank you very much for looking into this. In the meantime I tried some things, and I believe now that the reason was in the step that executes the pipeline, which is further below and cannot be seen in the photo. This was using a prefect task. I switched this to a non prefect function and now it seems to work. It works with both sequential and concurrent task runner. I am still trying to recreate the initial condition just to be sure.

Thanks for sharing all of that - it is much appreciated

Yes this looks good. I made a note in m source and next time I work on the sqlite part which I will soon I would love to swap it out. 🙂