Transform any push-based system into pull-based one. Basehook handles bufferization, and concurrency for you, so you can just start consuming payloads. It also offers a UI for inspecting and replaying events.
Thread-based grouping: Webhooks are grouped by thread ID. Configure JSON paths to extract thread IDs and revision numbers from any format (Slack, GitHub, Shopify, etc.).
Two consumption modes:
pop(only_last_revision=False)- Process updates one by one in orderpop(only_last_revision=True)- Buffer updates and only consume the latest revision, skipping outdated ones
Pull-based processing: Your application pulls updates via pop() instead of receiving direct webhook POSTs. Failed processing is marked as ERROR and visible in the UI for manual retry.
git clone https://github.com/mehdigmira/basehook.git
cd basehook
docker-compose up -dAccess at http://localhost:8000
One-click deploy with PostgreSQL auto-configured.
Visit the UI and create a webhook with thread ID and revision number paths.
pip install basehookimport asyncio
from basehook import Basehook
basehook = Basehook(database_url="postgresql+asyncpg://...")
webhook_name = ...
async def process_one():
# Process updates one by one (in order)
async with basehook.pop(webhook_name, only_last_revision=False) as update:
if update:
print(update)
async def process_last():
# Process last
async with basehook.pop(webhook_name, only_last_revision=True) as update:
if update:
print(update)
if __name__ == "__main__":
asyncio.run(process_one())MIT