philiprehberger-once 0.3.0

Ensure a function runs only once, regardless of how many times it's called.

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:

⭐ Star the repo

🐛 Report issues

💡 Suggest features

❤️ Sponsor development

🌐 All Open Source Projects

💻 GitHub Profile

🔗 LinkedIn Profile

License

MIT