#diezler

1 messages · Page 1 of 1 (latest)

long tokenBOT
bleak ridge
#

My webhook + models

@router.post("/webhook")
async def stripe_webhook(event: dict, request: Request, stripe_signature=Header(None)):
    webhook_secret = os.getenv("WEBHOOK_SECRET")
    raw_body = await request.body()
    try:
        event = stripe.Webhook.construct_event(
            raw_body, stripe_signature, webhook_secret
        )

    except ValueError as e:
        # Invalid payload
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid payload.")

    except stripe.error.SignatureVerificationError as e:
        # Invalid signature

        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid signature.")

    # Handle the checkout.session.completed event
    if event["type"] == "checkout.session.completed":
        session = event["data"]["object"]

        # Fulfill the purchase...
        handle_checkout_session(session)

    return "Success"

class StripeCustomer(Base):
    __tablename__ = "stripe_customers"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    user_id = Column(Integer, ForeignKey('users.id'))
    stripeCustomerId = Column(String(255), nullable=False)
    stripeSubscriptionId = Column(String(255), nullable=False)
    user = relationship('User', back_populates='stripe_customer')
# 
class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    google_sub = Column(String, unique=True, nullable=True)
    username = Column(String, unique=True)
    email = Column(String, unique=True)
    hashed_password = Column(String)
    stripe_customer = relationship('StripeCustomer', back_populates='user')
#

Where should i create the stripe user or fetch an existing user? And how can i pass it into the checkout and check if a user subscription is active?

fossil escarp
#

clueless how I would track the status of the subscription (creating stripe user / getting sub id)
What part of this are you having issues with specifically?

#

You configuration broadly seems to make sense, you'd track these new subscriptions using those checkout.session.completed events you're monitoring

#

Where should i create the stripe user or fetch an existing user?
For what purpose? WHat are you trying to do?

#

And how can i pass it into the checkout and check if a user subscription is active?
This suggests you're trying to create an additional subscription for an existing customer, which you'd provide while creating the checkout session using customer and the stripe customer ID

bleak ridge
#

I am clueless at which point I should create the stripe customer to listen for the subscription and how i later can check if a subscription is active or canceled

bleak ridge
fossil escarp
fossil escarp
#

how i later can check if a subscription is active or canceled
In what context? If you have the subscription ID you can retrieve it. If you have a customer ID, you can list their subscriptions and inspect those. WHat exactly are you trying to do?

You should be listening to webhooks to get notification about changes to subscriptions, though: https://stripe.com/docs/billing/subscriptions/webhooks

bleak ridge
#

I know the webhooks, but what would be the best way to check for the status. Lets say I have an endpoint and i want the user only be able to use it if their subscription is active

#

Could this work?(just an outline)

from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel
import stripe

app = FastAPI()

# Replace with your Stripe secret key
stripe.api_key = "your_stripe_secret_key"

class CreateUserRequest(BaseModel):
    username: str
    email: str

class SubscriptionStatusResponse(BaseModel):
    subscription_id: str
    status: str

@app.post("/create_user")
def create_user(user_data: CreateUserRequest):

    user_id = "123" 

    # Create a Stripe customer
    stripe_customer = stripe.Customer.create(
        email=user_data.email,
        name=user_data.username,
        description=f"App user ID: {user_id}"
    )

    link_stripe_customer_to_user(user_id, stripe_customer.id)

    return {"message": "User created successfully", "user_id": user_id}

def link_stripe_customer_to_user(user_id: str, stripe_customer_id: str):
    pass

@app.post("/check_subscription_status")
def check_subscription_status(user_id: str):
    # Retrieve the Stripe customer ID associated with the app user
    stripe_customer_id = get_stripe_customer_id(user_id)

    if stripe_customer_id is None:
        raise HTTPException(
            status_code=404,
            detail="Stripe customer not found for the given user ID"
        )

    # Retrieve the customer's subscriptions from Stripe
    subscriptions = stripe.Subscription.list(customer=stripe_customer_id)

    # For simplicity, we'll assume the user has only one subscription
    if subscriptions.data:
        subscription = subscriptions.data[0]
        return SubscriptionStatusResponse(
            subscription_id=subscription.id,
            status=subscription.status
        )
    else:
        return SubscriptionStatusResponse(
            subscription_id=None,
            status="No active subscription"
        )
    
def get_stripe_customer_id(user_id: str):
    return "stripe_customer_id_for_user_" + user_id
fossil escarp
#

It could, depend on what you need to check in that subscription list response, yea

#

that seems to be looking at only the first list result (data[0]) and using that status, so consider whether you may have multiple subscriptions and how you'd handle that

bleak ridge
#

each user can onyl have one sub

#

*only

fossil escarp
#

Ok, then sure

bleak ridge
#

Would it be unsafe to store the sub satus directly in the db, so I can avoid to fetch the stripe endpoint everytime? I would update the status with the webhook events.

fossil escarp
#

You could cache it locally, sure!