DEV Community: zwx00

Replay failed stripe events via webhook

zwx00 — Fri, 20 Dec 2024 10:55:58 +0000

Sometimes webhook events fail to deliver, and you need to replay them to ensure your system processes all important events. Here's a handy one-liner using the Stripe CLI to resend failed subscription cancellation events:

stripe events list \
  --type=customer.subscription.deleted \
  --delivery-success=false \
  --live \
  --limit 150 \
  | jq ".data[].id" \
  | xargs -n1 -t stripe events resend \
    --live \
    --webhook-endpoint=we_LALALALA

Let's break down what this command does:

stripe events list: Lists Stripe events
- --type=customer.subscription.deleted: Filters for subscription cancellation events
- --delivery-success=false: Only shows failed deliveries
- --live: Uses live mode (not test mode)
- --limit 150: Retrieves up to 150 events
jq ".data[].id": Extracts just the event IDs from the JSON response
xargs -n1 -t: Processes each event ID one at a time
- -n1: Passes one argument per command
- -t: Prints each command before executing it
stripe events resend: Resends each event to your webhook endpoint
- --live: Uses live mode
- --webhook-endpoint=we_LALALALA: Specifies the webhook endpoint to use

Remember to replace we_LALALALA with your actual webhook endpoint ID.

This command is particularly useful when:

Your webhook endpoint was down
You had network issues
You're testing new webhook handling code
You need to backfill missed events

Make sure you have both the Stripe CLI and jq installed before running this command.

Happy webhooking! 🎣

Async Axiom logging

zwx00 — Sun, 17 Mar 2024 20:07:59 +0000

Axiom Logger documentation for its python client does not provide any examples on how to log in async fashion. Both Axiom logger and the default python logger utility are blocking. And while streaming to stdout is not a big issue, blocking network responses might be (axiom-py utilizies raw urllib below the hood). Here's how we addressed this at Katalist.

We have a file called logging.py from where we import the logging handlers we're gonna be using with our logging config.

# main.py
# ... other imports
from katalist.logging import handlers

logging.basicConfig(
    handlers=handlers,
    force=True,
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",  # Define log message format.
    datefmt="%Y-%m-%d %H:%M:%S",  # Define the date format to include in log messages.
)

app = FastAPI()

# all the views...

One important thing to note here: without setting force to True, the handler configuration was being ignored. Other than that, there's nothing special about this setup.

So how do we include the AxiomHandler from axiom-py which is by nature synchronous, among our handlers? The answer lies in a QueueHandler utility that python's logging module provides by default. This allows us to basically enqueue logged entries that are then consumed by another thread, started by QueueListener. Python's queue, a thread-safe queue implementation takes care of the communication between threads.

# logging.py
import queue
from logging import Handler, LogRecord, StreamHandler
from logging.handlers import QueueHandler, QueueListener

from axiom.logging import AxiomHandler, Client

from katalystai.conf import SETTINGS


handlers: list[Handler] = [StreamHandler()]

if SETTINGS.mode == "production":

    log_queue = queue.Queue()
    queue_handler = QueueHandler(log_queue)

    axiom_client = Client(token=SETTINGS.axiom_token)

    axiom_handler = AxiomHandler(
        client=axiom_client,
        dataset="katalist-backend",
    )

    QueueListener(log_queue, axiom_handler).start()
    handlers.append(queue_handler)

This is already nice, but another issue remains. QueueHandler will strip off everything but the raw message to prevent pickling errors when passing messages to another thread. But it's 2024 and everyone knows structured logging is the name of the game. And Axiom itself is good for nothing if not for processing structured logs. Hence we need to modify how the QueueHandler processes our messages.

To achieve this we will subclass QueueHandler to keep the args parameter intact.

Here's the final code:


import queue
from logging import INFO, Handler, LogRecord, StreamHandler
from logging.handlers import QueueHandler, QueueListener

from axiom.logging import AxiomHandler, Client

from katalystai.conf import SETTINGS

handlers: list[Handler] = [StreamHandler()]


class KatalistQueueHandler(QueueHandler):
    def prepare(self, record: LogRecord):
        args = record.args
        super().prepare(record)
        record.args = args

        return record


if SETTINGS.mode == "production":

    log_queue = queue.Queue()
    queue_handler = KatalistQueueHandler(log_queue)

    axiom_client = Client(token=SETTINGS.axiom_token)

    axiom_handler = AxiomHandler(
        client=axiom_client,
        dataset="katalist-backend",
        level=INFO,
    )

    QueueListener(log_queue, axiom_handler).start()
    handlers.append(queue_handler)

Now we can log structured data anywhere in our application without worrying about blocking network calls.

from logging import getLogger

logger = logging.getLogger(__name__)

logger.info("I am logged", { "myname": "robby bobby" })

While this workaround works nicely, you now need to make sure to only log data types that can be pickled. In all honestly, I don't see why you'd log anything but dicts, strings and numbers, but who knows. You've been warned.

Validating a Supabase JWT locally with Python and FastAPI

zwx00 — Fri, 15 Mar 2024 17:06:36 +0000

Supabase is nice, but the functionality it provides for verifying tokens (auth.get_user) for some inexplicable reasons makes a round trip to their end, and this trip in fact takes quite a while (up to 600ms).

JWT tokens should by their very nature be verifiable locally, provided we know the secret. Luckily Supabase does expose this JWT secret under project settings.

Here's how a FastAPI dependency to get user email and verify their authenticity using the token alone might look like:

import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

from katalystai.conf import SETTINGS



async def get_user_email(
    res: Response,
    cred: HTTPAuthorizationCredentials = Depends(HTTPBearer(auto_error=False)),
):

    if cred is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Bearer authentication required",
            headers={"WWW-Authenticate": 'Bearer realm="auth_required"'},
        )

    try:
        jwt_result = jwt.decode(
            cred.credentials,
            SETTINGS.supabase_jwt_secret,
            audience="authenticated",
            algorithms=["HS256"],
        )

        return jwt_result["email"]
    except jwt.exceptions.PyJWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": 'Bearer realm="auth_required"'},
        )

We can then use this "middleware" like so:

from fastapi import Depends
from katalystai.auth import get_user_email

@router.post
async def create_thing(user_email: str = Depends(get_user_email)):
    pass

Supabase python client overall is not that nice, it's kinda slow and unreliable. I also suggest switching to another ORM and communicating with Postgres directly instead of using their community python client. I had luck using tortoise, which has excellent async support and clean Django-like API.

How to impersonate a user through Django management shell

zwx00 — Sat, 16 Apr 2022 11:17:09 +0000

In a hypothetical scenario where you have access to the underlying system, there has to be a way to impersonate any given user that signed up on for your Django application.

It is however not as trivial to achieve this. There are packages like django-hijack etc., which provide this functionality, but we'd like to achieve it without installing new packages or modifying existing code. Here's a simple, non-intrusive way to go about it on a live environment.

Approach

First, you need to login into your Django application with any other account that you have access to. Visit the cookies tab in your development tools and copy the sessionid cookie value. It should look something like this: wxc0ldhcis45md5hbr3l7r4pyhewo0mr.

Then, on the system where Django server is running, access the Django management shell:

python manage.py dbshell

Then do the following:

# Import the required interfaces
from django.contrib.sessions.models import Session
from boltobserver.users.models import User # This will be different for you, depending on where your User model is
from django.contrib.sessions.backends.db import SessionStore

# Find the user you wish to impersonate
u = User.objects.filter(email="someone@something.com").first()

# Find the session you are currently using in your browser
s = Session.objects.filter(session_key = "wxc0ldhcis45md5hbr3l7r4pyhewo0mr").first()

# And finally, modify the session by binding it to your target user
# _auth_user_backend might be different for you, check settings.AUTHENTICATION_BACKENDS for the right value
s2.session_data = SessionStore().encode({"_auth_user_id": str(u.id), "_auth_user_backend": "allauth.account.auth_backends.AuthenticationBackend", "_auth_user_hash": u.get_session_auth_hash()})
s2.save()

After refreshing the page, you should be logged in as your desired user.

Thanks for reading!