Skip to main content
KVStore provides a way to share small pieces of state across all your serverless runners. Similar to Vercel KV or Cloudflare Workers KV, it requires no setup and uses automatic authentication with your fal credentials.

​
API Reference

​
Initialize KVStore

from fal.toolkit.kv import KVStore

kv = KVStore("myapp")  # "myapp" is your namespace
Each namespace is isolated - different db_name values create separate key-value stores.

​
Methods

get(key: str) -> Optional[str] Retrieve a value from the store. Returns None if the key doesn’t exist. 
value = kv.get("my-key")
if value is None:
    print("Key not found")
set(key: str, value: str) -> None Store a string value. The value must be a string - use json.dumps() for objects. 
import json

# Store a string
kv.set("my-key", "my-value")

# Store an object as JSON
kv.set("config", json.dumps({"enabled": True, "limit": 100}))

​
Basic Example

import fal
import json
from fal.toolkit.kv import KVStore

class MyApp(fal.App):
    def setup(self):
        self.kv = KVStore("myapp")
    
    @fal.endpoint("/")
    def process(self, input: Input):
        # Check cache
        cache_key = f"result-{input.id}"
        cached = self.kv.get(cache_key)
        
        if cached:
            return json.loads(cached)
        
        # Compute and cache result
        result = self.expensive_operation(input)
        self.kv.set(cache_key, json.dumps(result))
        
        return result

​
Common Use Cases

​
Caching API Responses

Cache external API calls to reduce latency and costs: 
class MyApp(fal.App):
    def setup(self):
        self.kv = KVStore("api-cache")
    
    @fal.endpoint("/")
    def process(self, input: Input):
        cache_key = f"weather:{input.city}"
        
        # Check cache
        cached = self.kv.get(cache_key)
        if cached:
            return json.loads(cached)
        
        # Fetch from external API
        weather = requests.get(f"https://api.weather.com/{input.city}").json()
        self.kv.set(cache_key, json.dumps(weather))
        
        return weather

​
Caching Computation Results

Store results of expensive operations to avoid recomputation: 
class MyApp(fal.App):
    def setup(self):
        self.kv = KVStore("compute-cache")
    
    @fal.endpoint("/")
    def process(self, input: Input):
        cache_key = f"result:{input.video_id}"
        
        cached = self.kv.get(cache_key)
        if cached:
            return json.loads(cached)
        
        # Expensive video processing
        result = self.process_video(input.video_url)
        self.kv.set(cache_key, json.dumps(result))
        
        return result

​
Important Limitations

No Atomic OperationsKVStore does not support atomic increments, locks, or compare-and-swap operations. Race conditions are possible when multiple runners write to the same key simultaneously. This makes it unsuitable for:
  • Counters (use Redis INCR instead)
  • Locks or semaphores (use Redis or PostgreSQL)
  • Any operation requiring atomicity

​
Use PostgreSQL/database instead when you need:

  • Complex queries
  • Relational data
  • Strong consistency guarantees
  • Transactions

​
Use /data volume instead when you need:

  • Large files (>25 MB)
  • Model weights or datasets
  • File-based storage
See Use Persistent Storage for more information about the /data volume.