1. Models (backend/models.py)- Added PaymentMethodType enum (card, cash, bank_transfer, check)- Added stripe_customer_id column to User model- Created new PaymentMethod model with all fields specified in the plan2. Alembic Migration (backend/alembic/versions/add_payment_methods.py)- Creates payment_methods table- Adds stripe_customer_id to users table- Creates appropriate indexes3. API Endpoints (backend/server.py)Added 12 new endpoints:Member Endpoints:- GET /api/payment-methods - List user's payment methods- POST /api/payment-methods/setup-intent - Create Stripe SetupIntent- POST /api/payment-methods - Save payment method after setup- PUT /api/payment-methods/{id}/default - Set as default- DELETE /api/payment-methods/{id} - Remove payment methodAdmin Endpoints:- GET /api/admin/users/{user_id}/payment-methods - List user's methods (masked)- POST /api/admin/users/{user_id}/payment-methods/reveal - Reveal sensitive details (requires password)- POST /api/admin/users/{user_id}/payment-methods/setup-intent - Create SetupIntent for user- POST /api/admin/users/{user_id}/payment-methods - Save method on behalf- POST /api/admin/users/{user_id}/payment-methods/manual - Record manual method (cash/check)- PUT /api/admin/users/{user_id}/payment-methods/{id}/default - Set default- DELETE /api/admin/users/{user_id}/payment-methods/{id} - Delete method4. Permissions (backend/permissions_seed.py)Added 5 new permissions:- payment_methods.view- payment_methods.view_sensitive- payment_methods.create- payment_methods.delete- payment_methods.set_default

This commit is contained in:
Koncept Kit
2026-01-31 01:03:17 +07:00
parent 03e5dd8bda
commit 9754f2db6e
6 changed files with 896 additions and 1 deletions

689
server.py
View File

@@ -17,7 +17,7 @@ import csv
import io
from database import engine, get_db, Base
from models import User, Event, EventRSVP, UserStatus, UserRole, RSVPStatus, SubscriptionPlan, Subscription, SubscriptionStatus, StorageUsage, EventGallery, NewsletterArchive, FinancialReport, BylawsDocument, Permission, RolePermission, Role, UserInvitation, InvitationStatus, ImportJob, ImportJobStatus, ImportRollbackAudit, Donation, DonationType, DonationStatus, SystemSettings
from models import User, Event, EventRSVP, UserStatus, UserRole, RSVPStatus, SubscriptionPlan, Subscription, SubscriptionStatus, StorageUsage, EventGallery, NewsletterArchive, FinancialReport, BylawsDocument, Permission, RolePermission, Role, UserInvitation, InvitationStatus, ImportJob, ImportJobStatus, ImportRollbackAudit, Donation, DonationType, DonationStatus, SystemSettings, PaymentMethod, PaymentMethodType
from auth import (
get_password_hash,
verify_password,
@@ -6448,6 +6448,693 @@ async def create_donation_checkout(
logger.error(f"Error creating donation checkout: {str(e)}")
raise HTTPException(status_code=500, detail="Failed to create donation checkout")
# ============================================================
# Payment Method Management API Endpoints
# ============================================================
class PaymentMethodResponse(BaseModel):
id: str
card_brand: Optional[str] = None
card_last4: Optional[str] = None
card_exp_month: Optional[int] = None
card_exp_year: Optional[int] = None
card_funding: Optional[str] = None
payment_type: str
is_default: bool
is_manual: bool
manual_notes: Optional[str] = None
created_at: datetime
model_config = {"from_attributes": True}
class PaymentMethodSaveRequest(BaseModel):
stripe_payment_method_id: str
set_as_default: bool = False
class AdminManualPaymentMethodRequest(BaseModel):
payment_type: Literal["cash", "bank_transfer", "check"]
manual_notes: Optional[str] = None
set_as_default: bool = False
class AdminRevealRequest(BaseModel):
password: str
def get_or_create_stripe_customer(user: User, db: Session) -> str:
"""Get existing or create new Stripe customer for user."""
import stripe
# Get Stripe API key
stripe_key = get_setting(db, 'stripe_secret_key', decrypt=True)
if not stripe_key:
stripe_key = os.getenv("STRIPE_SECRET_KEY")
if not stripe_key:
raise HTTPException(status_code=500, detail="Stripe API key not configured")
stripe.api_key = stripe_key
if user.stripe_customer_id:
# Verify customer still exists in Stripe
try:
customer = stripe.Customer.retrieve(user.stripe_customer_id)
# Check if customer was deleted using getattr (Stripe SDK doesn't expose 'deleted' directly)
if getattr(customer, 'deleted', False) or customer.get('deleted', False):
# Customer was deleted, create a new one
user.stripe_customer_id = None
else:
return user.stripe_customer_id
except stripe.error.InvalidRequestError:
# Customer doesn't exist, create a new one
user.stripe_customer_id = None
except Exception as e:
logger.warning(f"Error retrieving Stripe customer {user.stripe_customer_id}: {str(e)}")
user.stripe_customer_id = None
# Create new Stripe customer
customer = stripe.Customer.create(
email=user.email,
name=f"{user.first_name} {user.last_name}",
metadata={"user_id": str(user.id)}
)
user.stripe_customer_id = customer.id
db.commit()
logger.info(f"Created Stripe customer {customer.id} for user {user.id}")
return customer.id
@api_router.get("/payment-methods")
async def list_payment_methods(
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db)
):
"""List current user's saved payment methods."""
methods = db.query(PaymentMethod).filter(
PaymentMethod.user_id == current_user.id,
PaymentMethod.is_active == True
).order_by(PaymentMethod.is_default.desc(), PaymentMethod.created_at.desc()).all()
return [{
"id": str(m.id),
"card_brand": m.card_brand,
"card_last4": m.card_last4,
"card_exp_month": m.card_exp_month,
"card_exp_year": m.card_exp_year,
"card_funding": m.card_funding,
"payment_type": m.payment_type.value,
"is_default": m.is_default,
"is_manual": m.is_manual,
"manual_notes": m.manual_notes if m.is_manual else None,
"created_at": m.created_at.isoformat()
} for m in methods]
@api_router.post("/payment-methods/setup-intent")
async def create_setup_intent(
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db)
):
"""Create a Stripe SetupIntent for adding a new payment method."""
import stripe
# Get or create Stripe customer
customer_id = get_or_create_stripe_customer(current_user, db)
# Get Stripe API key
stripe_key = get_setting(db, 'stripe_secret_key', decrypt=True)
if not stripe_key:
stripe_key = os.getenv("STRIPE_SECRET_KEY")
stripe.api_key = stripe_key
# Create SetupIntent
setup_intent = stripe.SetupIntent.create(
customer=customer_id,
payment_method_types=["card"],
metadata={"user_id": str(current_user.id)}
)
logger.info(f"Created SetupIntent for user {current_user.id}")
return {
"client_secret": setup_intent.client_secret,
"setup_intent_id": setup_intent.id
}
@api_router.post("/payment-methods")
async def save_payment_method(
request: PaymentMethodSaveRequest,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db)
):
"""Save a payment method after successful SetupIntent confirmation."""
import stripe
# Get Stripe API key
stripe_key = get_setting(db, 'stripe_secret_key', decrypt=True)
if not stripe_key:
stripe_key = os.getenv("STRIPE_SECRET_KEY")
stripe.api_key = stripe_key
# Refresh user from DB to get latest stripe_customer_id
db.refresh(current_user)
# Retrieve payment method from Stripe
try:
pm = stripe.PaymentMethod.retrieve(request.stripe_payment_method_id)
except stripe.error.InvalidRequestError as e:
logger.error(f"Invalid payment method ID: {request.stripe_payment_method_id}, error: {str(e)}")
raise HTTPException(status_code=400, detail="Invalid payment method ID")
# Verify ownership - payment method must be attached to user's customer
pm_customer = pm.customer if hasattr(pm, 'customer') else None
logger.info(f"Verifying PM ownership: pm.customer={pm_customer}, user.stripe_customer_id={current_user.stripe_customer_id}")
if not current_user.stripe_customer_id:
raise HTTPException(status_code=403, detail="User does not have a Stripe customer ID")
if not pm_customer:
raise HTTPException(status_code=403, detail="Payment method is not attached to any customer")
if pm_customer != current_user.stripe_customer_id:
raise HTTPException(status_code=403, detail="Payment method not owned by user")
# Check for duplicate
existing = db.query(PaymentMethod).filter(
PaymentMethod.stripe_payment_method_id == request.stripe_payment_method_id,
PaymentMethod.is_active == True
).first()
if existing:
raise HTTPException(status_code=400, detail="Payment method already saved")
# Handle default setting - unset others if setting this as default
if request.set_as_default:
db.query(PaymentMethod).filter(
PaymentMethod.user_id == current_user.id,
PaymentMethod.is_active == True
).update({"is_default": False})
# Extract card details
card = pm.card if pm.type == "card" else None
# Create payment method record
payment_method = PaymentMethod(
user_id=current_user.id,
stripe_payment_method_id=request.stripe_payment_method_id,
card_brand=card.brand if card else None,
card_last4=card.last4 if card else None,
card_exp_month=card.exp_month if card else None,
card_exp_year=card.exp_year if card else None,
card_funding=card.funding if card else None,
payment_type=PaymentMethodType.card,
is_default=request.set_as_default,
is_active=True,
is_manual=False
)
db.add(payment_method)
db.commit()
db.refresh(payment_method)
logger.info(f"Saved payment method {payment_method.id} for user {current_user.id}")
return {
"id": str(payment_method.id),
"card_brand": payment_method.card_brand,
"card_last4": payment_method.card_last4,
"card_exp_month": payment_method.card_exp_month,
"card_exp_year": payment_method.card_exp_year,
"is_default": payment_method.is_default,
"message": "Payment method saved successfully"
}
@api_router.put("/payment-methods/{payment_method_id}/default")
async def set_default_payment_method(
payment_method_id: str,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db)
):
"""Set a payment method as the default for auto-renewals."""
try:
pm_uuid = uuid.UUID(payment_method_id)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid payment method ID")
payment_method = db.query(PaymentMethod).filter(
PaymentMethod.id == pm_uuid,
PaymentMethod.user_id == current_user.id,
PaymentMethod.is_active == True
).first()
if not payment_method:
raise HTTPException(status_code=404, detail="Payment method not found")
# Unset all other defaults
db.query(PaymentMethod).filter(
PaymentMethod.user_id == current_user.id,
PaymentMethod.is_active == True
).update({"is_default": False})
# Set this one as default
payment_method.is_default = True
db.commit()
logger.info(f"Set default payment method {payment_method_id} for user {current_user.id}")
return {"message": "Default payment method updated", "id": str(payment_method.id)}
@api_router.delete("/payment-methods/{payment_method_id}")
async def delete_payment_method(
payment_method_id: str,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db)
):
"""Delete (soft-delete) a saved payment method."""
import stripe
try:
pm_uuid = uuid.UUID(payment_method_id)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid payment method ID")
payment_method = db.query(PaymentMethod).filter(
PaymentMethod.id == pm_uuid,
PaymentMethod.user_id == current_user.id,
PaymentMethod.is_active == True
).first()
if not payment_method:
raise HTTPException(status_code=404, detail="Payment method not found")
# Detach from Stripe if it's a Stripe payment method
if payment_method.stripe_payment_method_id:
try:
stripe_key = get_setting(db, 'stripe_secret_key', decrypt=True)
if not stripe_key:
stripe_key = os.getenv("STRIPE_SECRET_KEY")
stripe.api_key = stripe_key
stripe.PaymentMethod.detach(payment_method.stripe_payment_method_id)
logger.info(f"Detached Stripe payment method {payment_method.stripe_payment_method_id}")
except stripe.error.StripeError as e:
logger.warning(f"Failed to detach Stripe payment method: {str(e)}")
# Soft delete
payment_method.is_active = False
payment_method.is_default = False
db.commit()
logger.info(f"Deleted payment method {payment_method_id} for user {current_user.id}")
return {"message": "Payment method deleted"}
# ============================================================
# Admin Payment Method Management Endpoints
# ============================================================
@api_router.get("/admin/users/{user_id}/payment-methods")
async def admin_list_user_payment_methods(
user_id: str,
current_user: User = Depends(require_permission("payment_methods.view")),
db: Session = Depends(get_db)
):
"""Admin: List a user's payment methods (masked)."""
try:
user_uuid = uuid.UUID(user_id)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid user ID")
user = db.query(User).filter(User.id == user_uuid).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
methods = db.query(PaymentMethod).filter(
PaymentMethod.user_id == user_uuid,
PaymentMethod.is_active == True
).order_by(PaymentMethod.is_default.desc(), PaymentMethod.created_at.desc()).all()
return [{
"id": str(m.id),
"card_brand": m.card_brand,
"card_last4": m.card_last4,
"card_exp_month": m.card_exp_month,
"card_exp_year": m.card_exp_year,
"card_funding": m.card_funding,
"payment_type": m.payment_type.value,
"is_default": m.is_default,
"is_manual": m.is_manual,
"manual_notes": m.manual_notes if m.is_manual else None,
"created_at": m.created_at.isoformat(),
# Sensitive data masked
"stripe_payment_method_id": None
} for m in methods]
@api_router.post("/admin/users/{user_id}/payment-methods/reveal")
async def admin_reveal_payment_details(
user_id: str,
request: AdminRevealRequest,
current_user: User = Depends(require_permission("payment_methods.view_sensitive")),
db: Session = Depends(get_db)
):
"""Admin: Reveal full payment method details (requires password confirmation)."""
try:
user_uuid = uuid.UUID(user_id)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid user ID")
# Verify admin's password
if not verify_password(request.password, current_user.password_hash):
logger.warning(f"Admin {current_user.email} failed password verification for payment reveal")
raise HTTPException(status_code=401, detail="Invalid password")
user = db.query(User).filter(User.id == user_uuid).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
methods = db.query(PaymentMethod).filter(
PaymentMethod.user_id == user_uuid,
PaymentMethod.is_active == True
).order_by(PaymentMethod.is_default.desc(), PaymentMethod.created_at.desc()).all()
# Log sensitive access
logger.info(f"Admin {current_user.email} revealed payment details for user {user_id}")
return [{
"id": str(m.id),
"card_brand": m.card_brand,
"card_last4": m.card_last4,
"card_exp_month": m.card_exp_month,
"card_exp_year": m.card_exp_year,
"card_funding": m.card_funding,
"payment_type": m.payment_type.value,
"is_default": m.is_default,
"is_manual": m.is_manual,
"manual_notes": m.manual_notes,
"created_at": m.created_at.isoformat(),
"stripe_payment_method_id": m.stripe_payment_method_id
} for m in methods]
@api_router.post("/admin/users/{user_id}/payment-methods/setup-intent")
async def admin_create_setup_intent_for_user(
user_id: str,
current_user: User = Depends(require_permission("payment_methods.create")),
db: Session = Depends(get_db)
):
"""Admin: Create a SetupIntent for adding a card on behalf of a user."""
import stripe
try:
user_uuid = uuid.UUID(user_id)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid user ID")
user = db.query(User).filter(User.id == user_uuid).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
# Get or create Stripe customer for the target user
customer_id = get_or_create_stripe_customer(user, db)
# Get Stripe API key
stripe_key = get_setting(db, 'stripe_secret_key', decrypt=True)
if not stripe_key:
stripe_key = os.getenv("STRIPE_SECRET_KEY")
stripe.api_key = stripe_key
# Create SetupIntent
setup_intent = stripe.SetupIntent.create(
customer=customer_id,
payment_method_types=["card"],
metadata={
"user_id": str(user.id),
"created_by_admin": str(current_user.id)
}
)
logger.info(f"Admin {current_user.email} created SetupIntent for user {user_id}")
return {
"client_secret": setup_intent.client_secret,
"setup_intent_id": setup_intent.id
}
@api_router.post("/admin/users/{user_id}/payment-methods")
async def admin_save_payment_method_for_user(
user_id: str,
request: PaymentMethodSaveRequest,
current_user: User = Depends(require_permission("payment_methods.create")),
db: Session = Depends(get_db)
):
"""Admin: Save a payment method on behalf of a user."""
import stripe
try:
user_uuid = uuid.UUID(user_id)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid user ID")
user = db.query(User).filter(User.id == user_uuid).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
# Refresh user to get latest data
db.refresh(user)
# Get Stripe API key
stripe_key = get_setting(db, 'stripe_secret_key', decrypt=True)
if not stripe_key:
stripe_key = os.getenv("STRIPE_SECRET_KEY")
stripe.api_key = stripe_key
# Retrieve payment method from Stripe
try:
pm = stripe.PaymentMethod.retrieve(request.stripe_payment_method_id)
except stripe.error.InvalidRequestError as e:
logger.error(f"Invalid payment method ID: {request.stripe_payment_method_id}, error: {str(e)}")
raise HTTPException(status_code=400, detail="Invalid payment method ID")
# Verify ownership - payment method must be attached to user's customer
pm_customer = pm.customer if hasattr(pm, 'customer') else None
logger.info(f"Admin verifying PM ownership: pm.customer={pm_customer}, user.stripe_customer_id={user.stripe_customer_id}")
if not user.stripe_customer_id:
raise HTTPException(status_code=403, detail="User does not have a Stripe customer ID")
if not pm_customer:
raise HTTPException(status_code=403, detail="Payment method is not attached to any customer")
if pm_customer != user.stripe_customer_id:
raise HTTPException(status_code=403, detail="Payment method not attached to user's Stripe customer")
# Check for duplicate
existing = db.query(PaymentMethod).filter(
PaymentMethod.stripe_payment_method_id == request.stripe_payment_method_id,
PaymentMethod.is_active == True
).first()
if existing:
raise HTTPException(status_code=400, detail="Payment method already saved")
# Handle default setting
if request.set_as_default:
db.query(PaymentMethod).filter(
PaymentMethod.user_id == user.id,
PaymentMethod.is_active == True
).update({"is_default": False})
# Extract card details
card = pm.card if pm.type == "card" else None
# Create payment method record
payment_method = PaymentMethod(
user_id=user.id,
stripe_payment_method_id=request.stripe_payment_method_id,
card_brand=card.brand if card else None,
card_last4=card.last4 if card else None,
card_exp_month=card.exp_month if card else None,
card_exp_year=card.exp_year if card else None,
card_funding=card.funding if card else None,
payment_type=PaymentMethodType.card,
is_default=request.set_as_default,
is_active=True,
is_manual=False,
created_by=current_user.id
)
db.add(payment_method)
db.commit()
db.refresh(payment_method)
logger.info(f"Admin {current_user.email} saved payment method {payment_method.id} for user {user_id}")
return {
"id": str(payment_method.id),
"card_brand": payment_method.card_brand,
"card_last4": payment_method.card_last4,
"message": "Payment method saved successfully"
}
@api_router.post("/admin/users/{user_id}/payment-methods/manual")
async def admin_record_manual_payment_method(
user_id: str,
request: AdminManualPaymentMethodRequest,
current_user: User = Depends(require_permission("payment_methods.create")),
db: Session = Depends(get_db)
):
"""Admin: Record a manual payment method (cash, check, bank transfer)."""
try:
user_uuid = uuid.UUID(user_id)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid user ID")
user = db.query(User).filter(User.id == user_uuid).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
# Map payment type string to enum
payment_type_map = {
"cash": PaymentMethodType.cash,
"bank_transfer": PaymentMethodType.bank_transfer,
"check": PaymentMethodType.check
}
payment_type = payment_type_map.get(request.payment_type)
if not payment_type:
raise HTTPException(status_code=400, detail="Invalid payment type")
# Handle default setting
if request.set_as_default:
db.query(PaymentMethod).filter(
PaymentMethod.user_id == user.id,
PaymentMethod.is_active == True
).update({"is_default": False})
# Create manual payment method record
payment_method = PaymentMethod(
user_id=user.id,
stripe_payment_method_id=None,
payment_type=payment_type,
is_default=request.set_as_default,
is_active=True,
is_manual=True,
manual_notes=request.manual_notes,
created_by=current_user.id
)
db.add(payment_method)
db.commit()
db.refresh(payment_method)
logger.info(f"Admin {current_user.email} recorded manual payment method {payment_method.id} ({payment_type.value}) for user {user_id}")
return {
"id": str(payment_method.id),
"payment_type": payment_method.payment_type.value,
"message": "Manual payment method recorded successfully"
}
@api_router.put("/admin/users/{user_id}/payment-methods/{payment_method_id}/default")
async def admin_set_default_payment_method(
user_id: str,
payment_method_id: str,
current_user: User = Depends(require_permission("payment_methods.set_default")),
db: Session = Depends(get_db)
):
"""Admin: Set a user's payment method as default."""
try:
user_uuid = uuid.UUID(user_id)
pm_uuid = uuid.UUID(payment_method_id)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid ID format")
user = db.query(User).filter(User.id == user_uuid).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
payment_method = db.query(PaymentMethod).filter(
PaymentMethod.id == pm_uuid,
PaymentMethod.user_id == user_uuid,
PaymentMethod.is_active == True
).first()
if not payment_method:
raise HTTPException(status_code=404, detail="Payment method not found")
# Unset all other defaults
db.query(PaymentMethod).filter(
PaymentMethod.user_id == user_uuid,
PaymentMethod.is_active == True
).update({"is_default": False})
# Set this one as default
payment_method.is_default = True
db.commit()
logger.info(f"Admin {current_user.email} set default payment method {payment_method_id} for user {user_id}")
return {"message": "Default payment method updated", "id": str(payment_method.id)}
@api_router.delete("/admin/users/{user_id}/payment-methods/{payment_method_id}")
async def admin_delete_payment_method(
user_id: str,
payment_method_id: str,
current_user: User = Depends(require_permission("payment_methods.delete")),
db: Session = Depends(get_db)
):
"""Admin: Delete a user's payment method."""
import stripe
try:
user_uuid = uuid.UUID(user_id)
pm_uuid = uuid.UUID(payment_method_id)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid ID format")
user = db.query(User).filter(User.id == user_uuid).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
payment_method = db.query(PaymentMethod).filter(
PaymentMethod.id == pm_uuid,
PaymentMethod.user_id == user_uuid,
PaymentMethod.is_active == True
).first()
if not payment_method:
raise HTTPException(status_code=404, detail="Payment method not found")
# Detach from Stripe if it's a Stripe payment method
if payment_method.stripe_payment_method_id:
try:
stripe_key = get_setting(db, 'stripe_secret_key', decrypt=True)
if not stripe_key:
stripe_key = os.getenv("STRIPE_SECRET_KEY")
stripe.api_key = stripe_key
stripe.PaymentMethod.detach(payment_method.stripe_payment_method_id)
logger.info(f"Detached Stripe payment method {payment_method.stripe_payment_method_id}")
except stripe.error.StripeError as e:
logger.warning(f"Failed to detach Stripe payment method: {str(e)}")
# Soft delete
payment_method.is_active = False
payment_method.is_default = False
db.commit()
logger.info(f"Admin {current_user.email} deleted payment method {payment_method_id} for user {user_id}")
return {"message": "Payment method deleted"}
@api_router.post("/contact")
async def submit_contact_form(
request: ContactFormRequest,