Stripe webhook not working

So I need help as I am running this in a discord bot. and even though stripe says that it has processed the webhook events, code doesnt run.

code snippet:

def create_checkout_session(user_id, plan_name, success_url, cancel_url):
    price_id_map = {
        'starter': 'price_..........',  
        'premium': 'price_..........'  
    }
    
    session = stripe.checkout.Session.create(
        payment_method_types=['card'],
        line_items=[
            {
                'price': price_id_map[plan_name],
                'quantity': 1,
            },
        ],
        mode='subscription',
        metadata={
            'user_id': user_id,
            'plan': plan_name
        },
        success_url=success_url,
        cancel_url=cancel_url,
    )
    
    return session.url
app = Flask(__name__)
@app.route('/create-checkout-session')
def create_checkout():
    user_id = request.args.get('user_id') 
    plan_name = request.args.get('plan_name')  
    success_url = url_for('success', _external=True)
    cancel_url = url_for('cancel', _external=True)
    
    checkout_url = create_checkout_session(user_id, plan_name, success_url, cancel_url)
    
    return redirect(checkout_url)

@app.route('/success')
def success():
    return "Subscription successful!"

@app.route('/cancel')
def cancel():
    return "Subscription canceled."
endpoint_secret = "whsec_... ... ... " #I can fix this later.
@app.route('/webhook', methods=['POST'])
def stripe_webhook():
    payload = request.get_data(as_text=True)

    event_thread = threading.Thread(target=process_webhook_event, args=(payload))
    event_thread.start()

    return jsonify(success=True)
    
def process_webhook_event(payload):

    try:
        event = stripe.Event.construct_from(json.loads(payload), stripe.api_key)
    except ValueError as e:
        return 'Invalid payload', 400
    except stripe.error.SignatureVerificationError as e:
        return 'Invalid signature', 400

    if event.type == 'checkout.session.completed':
        session = event.data.object
        user_id = session.metadata.user_id
        plan = session.metadata.plan
        subscription_id = session.subscription

        if user_id and plan:
            try:
                user_data = y[0][user_id]['plans']
            except KeyError:
                y[0][user_id] = {'plans': []}
                user_data = y[0][user_id]['plans']
            user_data.append(plan)
            y[0][user_id]['subscription_id'] = subscription_id
            db.save_data()
            print(f"Added {plan} plan for user {user_id} with subscription ID {subscription_id}.")
        else:
            print(f"Missing user_id or plan in session metadata: {session.metadata}")

the reason why i cant send my bot’s code here is because it contains > 4000 lines of code. so it would be a mess to just comb through everything…

There is no error popping up by the way, I have inputted the stripe api key as well. ( This is in test mode )

But I guess you can just simulate a bot with no commands, just an on_ready event and a bot token run.

don’t worry, it doesn’t hurt to send a repl link for us to debug

try adding print statements to see which part of the code is being run.


Hi @Idkwhttph whats up?

2 Likes

i dont have a repl link, I dont host any projects on replit anymore. All my discord bots are hosted either on another website or VPS. i already added print statements for debugging, nothing printed out on the console. so i figured that the entire code wasnt working.

I dont have anymore code being hosted on replit anymore due to it being incompatible for my needs, and I dont want to make my code open sourced since I verified my discord bot already, and I dont want to have a bunch of copycats.

1 Like

Your error is with the threading threading.Thread(target=process_webhook_event, args=(payload)) since args expects a tuple. threading.Thread(target=process_webhook_event, args=(payload,)) is correct.

apologies, this doesnt work. but it doesnt present any error. it doesnt activate any debug prints at /webhook.

I have confirmed that the endpoint secret and all that works.

I checked stripe’s dashboard, it had indeed sent the checkout.session.completed.

1 Like

Why are you threading process_webhook_event to begin with?

Try to construct the webhook event directly in the webhook route like:

@app.route('/webhook', methods=['POST'])
def stripe_webhook():
    payload = request.get_data(as_text=True)
    sig_header = request.headers.get('Stripe-Signature')

    try:
        event = stripe.Webhook.construct_event(payload, sig_header, endpoint_secret)
    except ValueError as e:
        print(f"Error while processing webhook: {e}")
        return jsonify(success=False), 400
    except stripe.error.SignatureVerificationError as e:
        print(f"Signature verification failed: {e}")
        return jsonify(success=False), 400

    event_thread = threading.Thread(target=process_webhook_event, args=(event,))
    event_thread.start()

    return jsonify(success=True)

And, as already said add some print statements for debugging issues with payload processing or the signature verification

Yes, like you said, you can test the webhook with a simple:

import discord
import os

intents = discord.Intents.default()
client = discord.Client(intents=intents)

@client.event
async def on_ready():
    print(f'Logged in as {client.user}')

client.run('TOKEN')

Also, don’t forget that your Flask application must be accessible to Stripe’s webhook system. Which involves running your Flask app on a public server or using a tool like ngrok to expose your local server to the internet.

1 Like