https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> I am getting this error: ```ValueErr...
# ask-marvin
a
@Marvin I am getting this error:
Copy code
ValueError: Unable to create hash - objects could not be serialized.
  JSON error: Unable to serialize unknown type: <class 'sqlalchemy.engine.base.Engine'>
  Pickle error: cannot pickle '_thread.RLock' object

This often occurs when task inputs contain objects that cannot be cached like locks, file handles, or other system resources.

To resolve this, you can:
  1. Exclude these arguments by defining a custom `cache_key_fn`
  2. Disable caching by passing `cache_policy=NO_CACHE`
How can i fix this? running a self hosted prefect 3.7.0
m
✅ thought for 101.8 seconds
This happens because Prefect is trying to compute a cache key for a task call, and one of the task inputs is a SQLAlchemy 
Engine
. An 
Engine
contains internal resources like thread locks, so Prefect can't hash/serialize it: - JSON serialization fails for 
sqlalchemy.engine.base.Engine
- pickle/cloudpickle fails because it contains 
_thread.RLock
So the fix is usually one of these: 1. Don’t pass the 
Engine
into the task This is the cleanest option. Pass a connection URL or config into the task, and create the engine inside the task.
Copy code
python
from prefect import flow, task
from sqlalchemy import create_engine, text

@task
def run_query(db_url: str, user_id: int):
    engine = create_engine(db_url)
    try:
        with engine.connect() as conn:
            result = conn.execute(
                text("select * from users where id = :id"),
                {"id": user_id},
            )
            return [dict(row._mapping) for row in result]
    finally:
        engine.dispose()

@flow
def my_flow():
    return run_query("<postgresql+psycopg2://user:pass@host/db>", 123)
2. Exclude the engine from the cache key If you must pass the engine, tell Prefect not to use that parameter when computing the cache key.
Copy code
python
from prefect import task
from prefect.cache_policies import DEFAULT

@task(cache_policy=DEFAULT - "engine")
def run_query(engine, user_id: int):
    with engine.connect() as conn:
        result = conn.execute(...)
        return result.fetchall()
Or with a custom `cache_key_fn`:
Copy code
python
from prefect import task

def cache_without_engine(context, parameters):
    return f"user_id:{parameters['user_id']}"

@task(cache_key_fn=cache_without_engine)
def run_query(engine, user_id: int):
    with engine.connect() as conn:
        result = conn.execute(...)
        return result.fetchall()
3. Disable caching for that task entirely If caching is not important for this task:
Copy code
python
from prefect import task
from prefect.cache_policies import NO_CACHE

@task(cache_policy=NO_CACHE)
def run_query(engine, user_id: int):
    with engine.connect() as conn:
        result = conn.execute(...)
        return result.fetchall()
What I’d recommend For Prefect 3.7.0, best practice is: - pass simple, serializable inputs into tasks - create DB engines/sessions inside the task - avoid passing live connections, sessions, engines, locks, file handles, etc. between tasks If you're using SQLAlchemy sessions, the same advice applies: don’t pass a live 
Session
object as a task parameter. Docs: - Prefect caching docs - Advanced caching / non-serializable inputs - prefect-sqlalchemy integration If you want, paste the task signature that’s failing, and I can rewrite it into the safest Prefect 3 pattern.
a
@Marvin can i pass PREFECT_TASKS_DEFAULT_NO_CACHE=true into my worker environment variables instead?
m
✅ thought for 94.0 seconds
Yes — if your Prefect version includes that setting, setting 
PREFECT_TASKS_DEFAULT_NO_CACHE=true
on the worker can disable task caching by default for flows that worker imports/runs. But there are a couple important caveats: What it does - It changes the default task cache behavior to 
NO_CACHE
- That means Prefect won’t try to generate cache keys for tasks that are using the default cache behavior - So it can prevent the SQLAlchemy 
Engine
hashing error if that error is coming from default task caching Important caveats 1. It only affects tasks using the default cache behavior - If a task explicitly sets 
cache_key_fn=...
- or explicitly sets 
cache_policy=...
then this env var may not override that 2. Worker-side env matters - Since flows are imported/executed in the worker environment, setting it on the worker is the right place 3. This is a broad hammer - It disables default caching for all tasks on that worker - If you only have one or two failing tasks, it’s usually better to fix those tasks directly Best recommendation If this is just one task receiving a SQLAlchemy 
Engine
, I’d still recommend fixing the code instead of relying on a global setting: - create the engine inside the task, or - set that specific task to 
cache_policy=NO_CACHE
, or - exclude 
engine
from the cache key Example:
Copy code
python
from prefect import task
from prefect.cache_policies import NO_CACHE

@task(cache_policy=NO_CACHE)
def my_task(engine, query):
    ...
Or:
Copy code
python
from prefect import task
from prefect.cache_policies import DEFAULT

@task(cache_policy=DEFAULT - "engine")
def my_task(engine, query):
    ...
Bottom line - Yes, setting 
PREFECT_TASKS_DEFAULT_NO_CACHE=true
on the worker may solve it globally - But the safer, more targeted fix is to update the problematic task(s) If you want, I can help you check whether your current task is using default caching or explicit caching, and tell you whether this env var will actually fix your case.
2 Views
Open in Slack
PreviousNext
Powered by