#choz_best-practices

1 messages ¡ Page 1 of 1 (latest)

digital thistleBOT
#

👋 Welcome to your new thread!

⏲️ We'll be here soon! Typically we respond in a few minutes, but sometimes we might take a bit longer if the server is busy or if you have a particularly tricky question.

⏱️ We close idle threads, which makes them read-only. Once a thread is closed it won't be reopened, but you can always start a new thread if you have another question.

🔗 This thread will always be available, even after it's closed. You can find it again using Discord's search, or you can save this link: https://discord.com/channels/841573134531821608/1225937586242064405

📝 Have more to share? Add more details, code, screenshots, videos, etc. below.

quasi cipherBOT
mossy glacier
#

Hello! Checkout Sessions are designed to be used one time only, so generally there's not a lot of utility in storing them specifically.

#

If a returning Customer wants to pay again you should do what you're doing in the first scenario: create a new Checkout Session for that Customer and send them to it.

daring roost
mossy glacier
#

Can you tell me more about your overall flow? It sounds like it starts with a customer showing up and saving payment info for later use. What's the later use look like?

daring roost
#

It's a bit of a rabbit hole, but I'm gonna try to explain it in a way that cuts out any unnecessary details.

#

So, I'm currently designing a system for a business that can't use standard subscription models because they have variation on the days in which they withdraw fees (it's never a perfect month).

The flow starts with a customer signing up with our online form. Upon submission we check if the customer has a registered email or phone number in our database, and for added safety, we check if they are associated with a customer in Stripe.

# ...
def create_customer(name: dict, email: str, phone: str) -> str:
    # Check for existing customer in the database
    with g.db_conn as conn:
        with conn.cursor() as cursor:
            query = "SELECT 1 FROM public.customers WHERE email = %s OR phone = %s LIMIT 1"
            cursor.execute(query, (email, phone))
            if cursor.fetchone():
                return jsonify({'error': 'Customer with this email or phone already exists.'}), 409
    
    try:
        # Check for existing customer by email in Stripe
        existing_customers = stripe.Customer.list(email=email).data
        if existing_customers:
            logger.debug("Customer with given email already exists in Stripe.")
            return jsonify({'error': 'Customer with this email already exists in Stripe.'}), 409
    except stripe.StripeError as e:
        logger.error("Failed to query Stripe for existing customer.", exc_info=True)
        return jsonify({'error': 'Failed to communicate with Stripe.'}), 500

    try:
        customer = stripe.Customer.create(
            name=f"{name.get('lastname')}, {name.get('firstname')}",
            email=email,
            phone=phone,
        )
        return customer.id
    except stripe.StripeError:
        logger.error("Failed to create Stripe customer.", exc_info=True)
        return jsonify({'error': 'Failed to create customer in Stripe.'}), 500
# ...
#

Then this information is stored with a customer ID associated with it (I used this as the PK in the database since it's unique to the customer).

    response = create_customer(name, email, phone)
    if isinstance(response, (tuple, list)):  # Adjust based on how jsonify responses are structured
        return response  # This is a jsonify response object in case of an error

    customer_id = response
    try:
        with g.db_conn as conn:
            with conn.cursor() as cursor:
                query = """
                INSERT INTO public.customers (customer_id, email, phone, name, address, metadata)
                VALUES (%s, %s, %s, %s, %s, %s)
                """
                cursor.execute(query, (customer_id, email, phone, json.dumps(name), json.dumps(address), json.dumps(metadata)))
                conn.commit()
                return jsonify({'message': 'Application submitted successfully.', 'id': customer_id}), 201
    except IntegrityError as e:
        conn.rollback()
        stripe.Customer.delete(customer_id)
        return jsonify({'error': 'An error occurred. Please try again later.'}), 409
#

So this creates an account for our customer (with an associated customer object). When they're in their account portal, clicking a button generates a new checkout link for them with this.

def create_checkout_session(customer_id) -> stripe.Session:
    customer = stripe.Customer.retrieve(customer_id)
    session = stripe.checkout.Session.create(
        customer=customer,
        payment_method_types=['card', 'us_bank_account'],
        mode='setup',
        success_url=app.config['SUCCESS_URL']
    )
    return jsonify(session)
#

This checkout link however is only used to collect their payment method, which will later be used to manually assign a "Subscription" to the customer. I put subscription in quotes because it isn't a true Stripe Subscription, but an internal calendar-based charging system.

#

So I'm wondering if I should restrict the user from making repeated checkout sessions by holding onto the session ID they generated and using it to monitor what stage of the process they are in.

mossy glacier
#

That all makes sense to me so far. How do the "subscription" payments happen? Are you creating Payment Intents server-side while the customer is off-session?

mossy glacier
daring roost
daring roost
mossy glacier
#

How are you determining the default Payment Method for a Customer with multiple Payment Methods attached?

daring roost
#

I'm using this webhook currently:

    if event['type'] == 'setup_intent.succeeded':
        intent = event['data']['object']

        # Iterate to look for payment methods
        for payment_type in ['us_bank_account', 'card']:
            payment_methods = stripe.PaymentMethod.list(
                customer=intent['customer'],
                type=payment_type,
            )
            if len(payment_methods['data']) > 0:
                payment_method = payment_methods['data'][0]

                print(f"Setting default payment method to: {payment_method.id}")

                stripe.Customer.modify(
                    intent['customer'],
                    invoice_settings={
                        'default_payment_method': payment_method.id,
                    },
                )
                print("Default payment method set successfully.")
                break
        else:
            print('Could not detect payment method.')
mossy glacier
#

Ah, okay, so you're setting invoice_settings.default_payment_method. How are the payments made later?

daring roost
#

The payments are made later by the client checking and seeing what rate group the customer belongs in. I work for a union so the amounts charged are based off of salary, which the client has to check.

The client assigns the user's database entry a three digit code that is associated with this database table:

Charge Info Table

CREATE TABLE charge_info (
    type_code CHAR(3) PRIMARY KEY,
    data JSONB NOT NULL,
    CHECK (type_code ~ '^\d{3}$')
);
  • type_code: A primary key that uniquely identifies each type of charge. It is a three-digit code, constrained to exactly three numeric digits.
  • data: A JSONB column storing charge-related data. This can include various nested fields such as the amount, payment type, etc.
#

Then the backend calendar system queries the customer database and charges customers with the associated task_type.

CREATE TABLE scheduled_tasks (
    id SERIAL PRIMARY KEY,
    scheduled_time TIMESTAMP WITHOUT TIME ZONE NOT NULL,
    task_type CHAR(3) NOT NULL,
    status TEXT DEFAULT 'pending',
    attempt_count INTEGER DEFAULT 0,
    CONSTRAINT fk_task_type
        FOREIGN KEY(task_type) 
        REFERENCES charge_info(type_code),
    UNIQUE(scheduled_time, task_type)
);
  • id: A unique identifier for each scheduled task, auto-incremented.
  • scheduled_time: The timestamp when the task is scheduled to be executed. It does not include a time zone.
  • task_type: References the type_code from the charge_info table, indicating the type of charge this task relates to.
  • status: The current status of the task, defaulting to 'pending'. This field tracks the task's progress.
  • attempt_count: Counts the number of attempts made to process the task, starting at 0. It's used to limit retries.
  • The UNIQUE constraint on scheduled_time and task_type ensures that there are no duplicate tasks scheduled for the same time and type.
mossy glacier
#

How does the actual charge take place? Are you creating a Payment Intent?

daring roost
#

Yes, in charge_customer (rest is just there for context in case you need it). This is from a manual test build, not the calendar build itself:

import stripe
import logging
import os

from flask import Flask, jsonify, request

from missing_method_mail import send_missing_payment_method_email

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

app = Flask(__name__)

# Stripe API Keys
stripe.api_key = os.getenv('STRIPE_SECRET_KEY')

# This is a placeholder for customer IDs retrieved from database
customer_id_list = []

def retrieve(id: str) -> stripe.Customer | None:
    try:
        customer = stripe.Customer.retrieve(id)
        return customer
    except stripe.error.InvalidRequestError:
        logger.debug(f"Customer with ID {id} could not be retrieved.")
    except Exception as e:
        logger.debug(f"An unhandled error occurred processing customer with ID {id}: {str(e)}")
    return None

def grab_customers(id_list: list[str]) -> list[stripe.Customer]:
    customers = []
    for id in id_list:
        customer = retrieve(id)
        if customer:
            customers.append(customer)
    return customers

def get_customer_payment_method(customer: stripe.Customer) -> str | None:
    default_payment_method_id = customer.invoice_settings.default_payment_method
    
    if default_payment_method_id:
        payment_method = stripe.PaymentMethod.retrieve(default_payment_method_id)
        method = payment_method.type
        
        if method in ["card", "us_bank_account"]:
            return payment_method
        else:
            logger.debug(f"Invalid payment method type for customer with ID {customer.id}.")
    else:
        logger.debug(f"No default payment method set for customer with ID {customer.id}.")
    return None

def charge_customer(customer: stripe.Customer, payment_method: stripe.PaymentMethod, amount: int):
    try:
        payment_intent = stripe.PaymentIntent.create(
            amount=amount,
            currency='usd',
            customer=customer.id,
            payment_method=payment_method.id,
            off_session=True,
            confirm=True,
            payment_method_types=[payment_method.type],
        )
        logger.info(f"Payment Intent created successfully: {payment_intent.id} for customer {customer.id}")
    except stripe.error.StripeError as e:
        logger.error(f"An error occurred while creating a payment intent for customer {customer.id}: {e}")

def determine_charge_amount(payment_method: stripe.PaymentMethod) -> int:
    """Replace these numbers with values retrieved 
    from postgres database in the future"""
    amount = 2400
    if payment_method.type == "card":
        # 
        credit_card_upcharge = 100
        amount += credit_card_upcharge
    return amount

customers = grab_customers(customer_id_list)

for customer in customers:
    payment_method = get_customer_payment_method(customer)
    if payment_method:
        amount = determine_charge_amount(payment_method)
        charge_customer(customer, payment_method, amount)
    else:
        # Prepare info for the email
        info = {
            'customer_id': customer.id,
            'customer_name': customer.name or "Unknown",
            'customer_email': customer.email or "No email provided",
            'currency': 'USD'
        }
        company_email = os.getenv("COMPANY_EMAIL")
        send_missing_payment_method_email(company_email, info)
mossy glacier
#

Yep, that makes sense to me!

daring roost
#

Cool! By the way, I know it's not part of the initial question, but do you see any major issues with my flow? I just want to make sure this auto-charge system isn't violating TOS or if there's ways to make some of the logic easier with other Stripe features.

#

I know a lot of the stuff has built-in Stripe features that work for it, but a lot of it was done manually since there were some limitations regarding ACH payments.

mossy glacier
#

I can't speak to any TOS violations or anything like that here, for that you'd need to talk to Stripe support. From a technical standpoint it seems fine to me. It's not unusual for people to have their own subscription logic outside Stripe and implement it like you are.

daring roost
#

Cool! Thanks for the feedback :)

I think that's all I need for now, but I appreciate the help a ton!