philiprehberger-once
v0.3.0698 downloadsMIT
Ensure a function runs only once, regardless of how many times it's called
$
pip install philiprehberger-onceREADME
philiprehberger-once
Ensure a function runs only once, regardless of how many times it's called.
Installation
pip install philiprehberger-once
Usage
from philiprehberger_once import once
@once
def load_config():
print("Loading...")
return {"debug": True}
load_config() # prints "Loading...", returns {"debug": True}
load_config() # returns {"debug": True} without printing
Async support
import asyncio
from philiprehberger_once import once
@once
async def fetch_token():
print("Fetching...")
return "abc-123"
asyncio.run(fetch_token()) # prints "Fetching...", returns "abc-123"
asyncio.run(fetch_token()) # returns "abc-123" without fetching
Once per key
from philiprehberger_once import once_per_key
@once_per_key
def connect(host, port=5432):
print(f"Connecting to {host}...")
return f"conn:{host}"
connect("db-1") # prints "Connecting to db-1...", returns "conn:db-1"
connect("db-1") # returns cached "conn:db-1"
connect("db-2") # prints "Connecting to db-2...", returns "conn:db-2"
Once per key (async)
import asyncio
from philiprehberger_once import once_per_key_async
@once_per_key_async
async def fetch_user(user_id: str):
print(f"Fetching {user_id}...")
return {"id": user_id}
async def main():
# Concurrent awaiters of the same key share one in-flight execution.
a, b, c = await asyncio.gather(
fetch_user("u1"),
fetch_user("u1"),
fetch_user("u2"),
)
# Only two prints: one for "u1" and one for "u2".
asyncio.run(main())
# Derive the cache key from a callable instead of the first arg:
@once_per_key_async(key=lambda req, **kw: kw["uid"])
async def load(req, *, uid: str):
return uid
Once per args
Use the entire call signature (positional + keyword) as the cache key — handy when the value depends on more than just the first argument.
from philiprehberger_once import once_per_args
@once_per_args
def init(host: str, port: int):
print(f"connecting to {host}:{port}")
return f"{host}:{port}"
init("a", 1) # prints, returns "a:1"
init("a", 1) # cached
init("a", 2) # prints, returns "a:2"
Reset and inspect
from philiprehberger_once import once
@once
def init():
return 42
init()
init.called # True
init.reset()
init.called # False
init() # runs again
API
| Function / Property | Description |
|---|---|
once(fn) | Decorator. Runs fn once, caches and returns the result on subsequent calls. Thread-safe. Supports async. |
once_per_key(fn) | Decorator. Runs fn once per unique first argument. Thread-safe. |
once_per_args(fn) | Decorator. Runs fn once per unique combination of positional and keyword arguments. Thread-safe. All arguments must be hashable. |
once_per_key_async(fn=None, *, key=None) | Decorator for async functions. Runs the coroutine once per unique key (first positional arg, or derived via key=...). Concurrent awaiters of the same key share one in-flight execution. |
.called | bool for once, dict[key, bool] for once_per_key and once_per_key_async. Whether the function has been called. |
.reset() | Clear cached result so the function can run again. once_per_key and once_per_key_async accept an optional key argument. |
Development
pip install -e .
python -m pytest tests/ -v
Support
If you find this project useful: