<@ULVA73B9P> in kubernetes self hosted prefect I w...
# ask-marvinr
Russell Brooks
05/07/2026, 2:23 PM@Marvin in kubernetes self hosted prefect I want to reuse the same redis app as generated by the prefect-helm chart for purposes of result storage and cache. Can you give a small example of how to set that up? Presumably I'll need to set up a Block? What else?
m
Marvin
05/07/2026, 2:23 PM❌ Error: unhandled errors in a TaskGroup (1 sub-exception)
Marvin
05/07/2026, 2:23 PMSorry, I encountered an error while processing your request. Please try again.
r
Russell Brooks
05/07/2026, 2:25 PM@Marvin can you explain and show an example of how to use the redis sub chart from prefect-helm as a result storage and for cache?
m
Marvin
05/07/2026, 2:25 PM✅ thought for 208.3 seconds
Marvin
05/07/2026, 2:28 PMYep — you can use the Redis instance created by the
prefect-helmprefect-helmprefect-helm Copy code
yaml
redis:
enabled: true
architecture: standalone
auth:
password: "my-redis-password" Copy code
yaml
backgroundServices:
runAsSeparateDeployment: true
messaging:
broker: prefect_redis.messaging
cache: prefect_redis.messaging Copy code
text
<helm-release-name>-redis-masterprefect Copy code
text
prefect-redis-master Copy code
text
prefect-redis-master.<namespace>.svc.cluster.localRedisStorageContainerhostportdbusernamepasswordconnection_stringresult_storage Copy code
python
from prefect.blocks.redis import RedisStorageContainer
from pydantic import SecretStr
redis_storage = RedisStorageContainer(
host="prefect-redis-master",
port=6379,
db=1,
password=SecretStr("my-redis-password"),
)
redis_storage.save("helm-redis-results", overwrite=True)db=0db=1db=2 Copy code
python
from prefect import flow, task
from prefect.blocks.redis import RedisStorageContainer
from pydantic import SecretStr
redis_storage = RedisStorageContainer(
host="prefect-redis-master",
port=6379,
db=1,
password=SecretStr("my-redis-password"),
)
@task(persist_result=True)
def compute(x: int) -> int:
print(f"running compute({x})")
return x * 10
@flow(persist_result=True, result_storage=redis_storage)
def my_flow(x: int) -> int:
return compute(x)
if __name__ == "__main__":
print(my_flow(5))Marvin
05/07/2026, 2:28 PM- returning the saved result instead of re-running the task
So task caching depends on persisted results. If you want Redis-backed cache behavior, use:
-
persist_result=TrueINPUTS Copy code
python
from prefect import flow, task
from prefect.cache_policies import INPUTS
from prefect.blocks.redis import RedisStorageContainer
from pydantic import SecretStr
redis_storage = RedisStorageContainer(
host="prefect-redis-master",
port=6379,
db=1,
password=SecretStr("my-redis-password"),
)
@task(
persist_result=True,
cache_policy=INPUTS,
result_storage=redis_storage,
)
def expensive_calculation(x: int) -> int:
print(f"actually computing for {x}")
return x * 100
@flow
def cached_flow():
a = expensive_calculation(7)
b = expensive_calculation(7)
c = expensive_calculation(8)
return a, b, c
if __name__ == "__main__":
print(cached_flow())778 Copy code
python
from prefect.blocks.redis import RedisStorageContainer
from pydantic import SecretStr
RedisStorageContainer(
host="prefect-redis-master",
port=6379,
db=1,
password=SecretStr("my-redis-password"),
).save("helm-redis-results", overwrite=True) Copy code
python
from prefect import flow, task
from prefect.cache_policies import INPUTS
from prefect.blocks.redis import RedisStorageContainer
redis_storage = RedisStorageContainer.load("helm-redis-results")
@task(
persist_result=True,
cache_policy=INPUTS,
result_storage=redis_storage,
)
def transform(x: int) -> int:
print(f"transforming {x}")
return x + 1
@flow(persist_result=True, result_storage=redis_storage)
def my_flow():
return transform(3)prefect-redisprefect-redisRedisDatabaseprefect.blocks.redis.RedisStorageContainerprefect-redisprefect-redis-masterpersist_result=TrueMarvin
05/07/2026, 2:28 PMport=6379,
db=1,
password=SecretStr("my-redis-password"),
)
@task(
persist_result=True,
cache_policy=INPUTS,
result_storage=redis_storage,
)
def slow_add(x: int, y: int) -> int:
print(f"computing {x} + {y}")
return x + y
@flow(persist_result=True, result_storage=redis_storage)
def demo():
r1 = slow_add(1, 2)
r2 = slow_add(1, 2)
r3 = slow_add(2, 3)
return r1, r2, r3
if name == "__main__":
print(demo())
```
## 13. Docs / references
Useful links:
- Prefect Helm chart repo: prefect-helm
- Prefect Helm docs: prefect-server chart
- Prefect Redis collection: prefect-redis
- Prefect result persistence docs: results documentation
- Prefect caching docs: task caching documentation
If you want, I can also show:
- a Helm values.yaml + deployment example for this setup, or
- a version that uses Kubernetes secrets/env vars instead of hardcoding the Redis password.
r
Russell Brooks
05/07/2026, 2:40 PM@Marvin I'm confused between the redisdatabase block and the redisstoragecontainer block you mention. Also I only see redisdatabase in my ui.
m
Marvin
05/07/2026, 2:40 PM✅ thought for 155.8 seconds
Marvin
05/07/2026, 2:43 PMYes — that makes sense, and the distinction is a bit confusing.
Short answer:
-
RedisStorageContainerRedisDatabaseprefect-redisRedisDatabaseRedisStorageContainerRedisStorageContainer Copy code
python
from prefect.blocks.redis import RedisStorageContainerresult_storage=... Copy code
python
@flow(result_storage=redis_storage, persist_result=True)
def my_flow():
...RedisDatabase Copy code
python
from prefect_redis.blocks import RedisDatabaseset/getRedisDatabaseprefect-redisRedisDatabaseRedisStorageContainerRedisStorageContainerRedisDatabase“use Redis as Prefect result storage and cache backend”then
RedisStorageContainerRedisStorageContainer Copy code
python
from prefect import flow, task
from prefect.cache_policies import INPUTS
from prefect.blocks.redis import RedisStorageContainer
from pydantic import SecretStr
redis_storage = RedisStorageContainer(
host="prefect-redis-master",
port=6379,
db=1,
password=SecretStr("my-redis-password"),
)
@task(
persist_result=True,
cache_policy=INPUTS,
result_storage=redis_storage,
)
def compute(x: int) -> int:
print(f"computing {x}")
return x * 2
@flow(persist_result=True, result_storage=redis_storage)
def my_flow():
return compute(5)RedisStorageContainer Copy code
python
from prefect.blocks.redis import RedisStorageContainer Copy code
bash
prefect block register -f register_redis.pyRedisStorageContainerRedisStorageContainerMarvin
05/07/2026, 2:43 PM- you just want Redis-backed result storage/caching working
### Option B: register
RedisStorageContainerRedisDatabaseRedisDatabaseRedisStorageContainerRedisStorageContainerresult_storage=RedisStorageContainerget_client()RedisDatabaseRedisDatabaseRedisStorageContainerr
Russell Brooks
05/07/2026, 3:08 PM@Marvin I struggle with the host name. Which command can I run in the redis pod to find the fully qualified domain name?
m
Marvin
05/07/2026, 3:08 PM✅ thought for 12.9 seconds
Marvin
05/07/2026, 3:09 PMYou usually don’t need to exec into the Redis pod to discover the FQDN — the fully qualified DNS name in Kubernetes is based on the Service, not the Pod.
For a Redis service named
prefect-redis-masterprefect Copy code
text
prefect-redis-master.prefect.svc.cluster.local Copy code
text
<service-name>.<namespace>.svc.cluster.local Copy code
bash
kubectl get svc -n <namespace> Copy code
bash
kubectl get svc -n prefect Copy code
text
<service-name>.prefect.svc.cluster.local Copy code
bash
cat /etc/resolv.conf Copy code
bash
kubectl exec -it -n prefect <redis-pod-name> -- cat /etc/resolv.conf Copy code
text
search prefect.svc.cluster.local svc.cluster.local cluster.local Copy code
bash
hostname -f Copy code
bash
nslookup <service-name> Copy code
bash
getent hosts <service-name> Copy code
bash
kubectl exec -it -n prefect <redis-pod-name> -- nslookup prefect-redis-masterhostname -f Copy code
bash
kubectl get pods -A | grep redis Copy code
bash
kubectl get svc -n <namespace> Copy code
text
<service>.<namespace>.svc.cluster.localprefectprefect-redis-master Copy code
text
prefect-redis-master.prefect.svc.cluster.local Copy code
text
prefect-redis-master Copy code
text
prefect-redis-master.prefectprefect-helmr
Russell Brooks
05/07/2026, 3:31 PM@Marvin ok. And how to add a time to live for the result?
m
Marvin
05/07/2026, 3:31 PM❌ Error: unhandled errors in a TaskGroup (1 sub-exception)
Marvin
05/07/2026, 3:31 PMSorry, I encountered an error while processing your request. Please try again.
r
Russell Brooks
05/07/2026, 3:33 PM@Marvin for the redisstoragecontainer to be used for prefect result storage, how can I add a ttl, also known as time to live?
m
Marvin
05/07/2026, 3:33 PM✅ thought for 126.9 seconds
Marvin
05/07/2026, 3:35 PMShort answer: you can’t set a Redis TTL directly on
RedisStorageContainerRedisStorageContainer.write_path(...)SETttl=ex= Copy code
python
from datetime import timedelta
from prefect import task
@task(
persist_result=True,
cache_expiration=timedelta(hours=1),
)
def my_task():
... Copy code
python
from datetime import timedelta
from prefect import flow, task
from prefect.cache_policies import INPUTS
from prefect.blocks.redis import RedisStorageContainer
from pydantic import SecretStr
redis_storage = RedisStorageContainer(
host="prefect-redis-master.prefect.svc.cluster.local",
port=6379,
db=1,
password=SecretStr("my-redis-password"),
)
@task(
persist_result=True,
result_storage=redis_storage,
cache_policy=INPUTS,
cache_expiration=timedelta(hours=1),
)
def expensive_task(x: int) -> int:
print(f"computing {x}")
return x * 10
@flow
def my_flow():
a = expensive_task(5)
b = expensive_task(5)
return a, bRedisStorageContainerwrite_pathawrite_pathSETex=<seconds> Copy code
python
from pathlib import Path
import redis
from prefect.blocks.redis import RedisStorageContainer
class TTLRedisStorageContainer(RedisStorageContainer):
ttl_seconds: int = 3600
def write_path(self, path: Path | str, content: bytes) -> bool:
client = redis.Redis(
host=self.host,
port=self.port,
db=self.db,
username=self.username.get_secret_value() if self.username else None,
password=self.password.get_secret_value() if self.password else None,
)
try:
return client.set(str(path), content, ex=self.ttl_seconds)
finally:
client.close()EXPIREcache_expirationcache_expirationresult_storage=RedisStorageContainer(...)persist_result=Truecache_policy=INPUTScache_expiration=timedelta(...)Marvin
05/07/2026, 3:35 PMIf you want, I can show a production-ready custom
TTLRedisStorageContainerr
Russell Brooks
05/07/2026, 3:38 PM@Marvin yes make a production ready TTLRedisStorageContainer please. I can imagine everyone user would like the choice to add ttl otherwise redis will get too full.
m
Marvin
05/07/2026, 3:38 PM✅ thought for 156.3 seconds
Marvin
05/07/2026, 3:41 PMYes — and I should give you one with an important warning up front:
Adding Redis TTL at the storage layer is useful, but it changes failure behavior.
If a result expires in Redis before Prefect reads it back, then Prefect may no longer be able to retrieve that result. That is fine if you want ephemeral cache-like storage, but it is not ideal for durable flow results you may inspect later in the UI.
So the safest production recommendation is:
- use
cache_expirationRedisStorageContainerdefault_ttl_secondsSET ... EX <ttl>result_storageTTLRedisStorageContainerMarvin
05/07/2026, 3:41 PMasync with self._client() as client:
return await client.set(
str(path),
content,
ex=self.default_ttl_seconds,
)
@async_dispatch(awrite_path)
def write_path(self, path: Path | str, content: bytes) -> bool:
"""
Synchronous write with TTL if configured.
"""
client = self._get_sync_client()
try:
return client.set(
str(path),
content,
ex=self.default_ttl_seconds,
)
finally:
client.close()
async def aread_path(self, path: Path | str) -> bytes | None:
"""
Read bytes from Redis.
Returns None if the key does not exist, which may mean:
- the key never existed
- the key expired due to TTL
- the key was deleted externally
"""
async with self._client() as client:
content = await client.get(str(path))
if content is None and self.log_missing_keys:
logger.info(
"Redis key '%s' was not found. It may have expired due to TTL.",
path,
)
return content
@async_dispatch(aread_path)
def read_path(self, path: Path | str) -> bytes | None:
"""
Synchronous read.
"""
client = self._get_sync_client()
try:
content = client.get(str(path))
finally:
client.close()
if content is None and self.log_missing_keys:
logger.info(
"Redis key '%s' was not found. It may have expired due to TTL.",
path,
)
return content
@classmethod
def from_host(
cls,
host: str,
port: int = 6379,
db: int = 0,
username: Optional[str | SecretStr] = None,
password: Optional[str | SecretStr] = None,
default_ttl_seconds: Optional[int] = None,
log_missing_keys: bool = True,
) -> "TTLRedisStorageContainer":
if isinstance(username, str):
username = SecretStr(username)
if isinstance(password, str):
password = SecretStr(password)
return cls(
host=host,
port=port,
db=db,
username=username,
password=password,
default_ttl_seconds=default_ttl_seconds,
log_missing_keys=log_missing_keys,
)
@classmethod
def from_connection_string(
cls,
connection_string: str | SecretStr,
default_ttl_seconds: Optional[int] = None,
log_missing_keys: bool = True,
) -> "TTLRedisStorageContainer":
if isinstance(connection_string, str):
connection_string = SecretStr(connection_string)
return cls(
connection_string=connection_string,
default_ttl_seconds=default_ttl_seconds,
log_missing_keys=log_missing_keys,
)
Copy code
## How to create and save the block
Example:
```python
from pydantic import SecretStr
from custom_redis_storage import TTLRedisStorageContainer
block = TTLRedisStorageContainer.from_host(
host="prefect-redis-master.prefect.svc.cluster.local",
port=6379,
db=1,
password=SecretStr("my-redis-password"),
default_ttl_seconds=7 * 24 * 60 * 60, # 7 days
)
block.save("redis-results-ttl-7d", overwrite=True)Marvin
05/07/2026, 3:41 PMb = expensive_task(5)
return a, b
Copy code
## Recommended production settings
I would *not* default this to a tiny TTL.
Good starting points:
- *1 day* for very ephemeral cache-style workloads
- *7 days* for most result browsing / retries / debugging
- *14-30 days* if you use Redis results operationally and still want cleanup
Example:
```python
default_ttl_seconds=604800 # 7 dayscache_expirationcache_expiration Copy code
python
from datetime import timedelta
from prefect import task
from prefect.cache_policies import INPUTS
@task(
persist_result=True,
cache_policy=INPUTS,
cache_expiration=timedelta(days=7),
result_storage=redis_storage,
)
def my_task(x: int) -> int:
return x * 2custom_redis_storage.pyprefectredis Copy code
python
from custom_redis_storage import TTLRedisStorageContainer
# importing the class and saving a block document is usually enough
block = TTLRedisStorageContainer.from_connection_string(
"redis://:my-redis-password@prefect-redis-master.prefect.svc.cluster.local:6379/1",
default_ttl_seconds=604800,
)
block.save("redis-results-ttl-7d", overwrite=True)4 Views