From 4dc9d7d5a53b4c9b8d49233cf2c384dda35c5313 Mon Sep 17 00:00:00 2001 From: Ben Sima Date: Sun, 9 Nov 2025 16:21:55 -0500 Subject: feat: Add stripe to Python deps and document dependency process - Add stripe to Omni/Bild/Deps/Python.nix (alphabetically sorted) - Fix all type annotations in Billing.py for mypy - Document how to add Python packages in AGENTS.md - Add billing routes to Web.py (checkout, portal, webhook) This enables Stripe integration in PodcastItLater. Related to task t-144e7lF --- Biz/PodcastItLater/Billing.py | 37 ++++++++++++++++++++----------------- 1 file changed, 20 insertions(+), 17 deletions(-) (limited to 'Biz/PodcastItLater/Billing.py') diff --git a/Biz/PodcastItLater/Billing.py b/Biz/PodcastItLater/Billing.py index e472889..025f1aa 100644 --- a/Biz/PodcastItLater/Billing.py +++ b/Biz/PodcastItLater/Billing.py @@ -12,6 +12,7 @@ import Biz.PodcastItLater.Core as Core import Omni.Log as Log import os import stripe +import typing from datetime import datetime from datetime import timezone @@ -54,7 +55,9 @@ PRICE_MAP = { } -def get_period_boundaries(user: dict) -> tuple[datetime, datetime]: +def get_period_boundaries( + user: dict[str, typing.Any], +) -> tuple[datetime, datetime]: """Get billing period boundaries for user. For paid users: use Stripe subscription period. @@ -90,7 +93,7 @@ def get_usage( user_id: int, period_start: datetime, period_end: datetime, -) -> dict: +) -> dict[str, int]: """Get usage stats for user in billing period. Returns: @@ -99,7 +102,7 @@ def get_usage( return Core.Database.get_usage(user_id, period_start, period_end) -def can_submit(user_id: int) -> tuple[bool, str, dict]: +def can_submit(user_id: int) -> tuple[bool, str, dict[str, int]]: """Check if user can submit article based on tier limits. Returns: @@ -119,7 +122,7 @@ def can_submit(user_id: int) -> tuple[bool, str, dict]: usage = get_usage(user_id, period_start, period_end) # Check article limit - article_limit = limits["articles_per_period"] + article_limit = limits["articles_per_period"] # type: ignore[index] if article_limit is not None and usage["articles"] >= article_limit: msg = ( f"You've reached your limit of {article_limit} articles " @@ -128,7 +131,7 @@ def can_submit(user_id: int) -> tuple[bool, str, dict]: return (False, msg, usage) # Check minutes limit (if implemented) - minute_limit = limits.get("minutes_per_period") + minute_limit = limits.get("minutes_per_period") # type: ignore[attr-defined] if minute_limit is not None and usage.get("minutes", 0) >= minute_limit: return ( False, @@ -185,7 +188,7 @@ def create_checkout_session(user_id: int, tier: str, base_url: str) -> str: else: session_params["customer_email"] = user["email"] - session = stripe.checkout.Session.create(**session_params) + session = stripe.checkout.Session.create(**session_params) # type: ignore[arg-type] logger.info( "Created checkout session for user %s, tier %s: %s", @@ -194,7 +197,7 @@ def create_checkout_session(user_id: int, tier: str, base_url: str) -> str: session.id, ) - return session.url + return session.url # type: ignore[return-value] def create_portal_session(user_id: int, base_url: str) -> str: @@ -229,7 +232,7 @@ def create_portal_session(user_id: int, base_url: str) -> str: return session.url -def handle_webhook_event(payload: bytes, sig_header: str) -> dict: +def handle_webhook_event(payload: bytes, sig_header: str) -> dict[str, str]: """Verify and process Stripe webhook event. Args: @@ -243,7 +246,7 @@ def handle_webhook_event(payload: bytes, sig_header: str) -> dict: May raise stripe.error.SignatureVerificationError if invalid signature """ # Verify webhook signature - event = stripe.Webhook.construct_event( + event = stripe.Webhook.construct_event( # type: ignore[no-untyped-call] payload, sig_header, STRIPE_WEBHOOK_SECRET, @@ -284,7 +287,7 @@ def handle_webhook_event(payload: bytes, sig_header: str) -> dict: return {"status": "processed", "type": event_type} -def _handle_checkout_completed(session: dict) -> None: +def _handle_checkout_completed(session: dict[str, typing.Any]) -> None: """Handle checkout.session.completed event.""" user_id = int(session.get("client_reference_id", 0)) customer_id = session.get("customer") @@ -301,17 +304,17 @@ def _handle_checkout_completed(session: dict) -> None: logger.info("Linked user %s to Stripe customer %s", user_id, customer_id) -def _handle_subscription_created(subscription: dict) -> None: +def _handle_subscription_created(subscription: dict[str, typing.Any]) -> None: """Handle customer.subscription.created event.""" _update_subscription_state(subscription) -def _handle_subscription_updated(subscription: dict) -> None: +def _handle_subscription_updated(subscription: dict[str, typing.Any]) -> None: """Handle customer.subscription.updated event.""" _update_subscription_state(subscription) -def _handle_subscription_deleted(subscription: dict) -> None: +def _handle_subscription_deleted(subscription: dict[str, typing.Any]) -> None: """Handle customer.subscription.deleted event.""" customer_id = subscription["customer"] @@ -326,7 +329,7 @@ def _handle_subscription_deleted(subscription: dict) -> None: logger.info("Downgraded user %s to free tier", user["id"]) -def _handle_payment_failed(invoice: dict) -> None: +def _handle_payment_failed(invoice: dict[str, typing.Any]) -> None: """Handle invoice.payment_failed event.""" customer_id = invoice["customer"] subscription_id = invoice.get("subscription") @@ -347,7 +350,7 @@ def _handle_payment_failed(invoice: dict) -> None: ) -def _update_subscription_state(subscription: dict) -> None: +def _update_subscription_state(subscription: dict[str, typing.Any]) -> None: """Update user subscription state from Stripe subscription object.""" customer_id = subscription["customer"] subscription_id = subscription["id"] @@ -393,7 +396,7 @@ def _update_subscription_state(subscription: dict) -> None: ) -def get_tier_info(tier: str) -> dict: +def get_tier_info(tier: str) -> dict[str, typing.Any]: """Get tier information for display. Returns: @@ -419,4 +422,4 @@ def get_tier_info(tier: str) -> dict: "description": "Unlimited articles", }, } - return tier_info.get(tier, tier_info["free"]) + return tier_info.get(tier, tier_info["free"]) # type: ignore[return-value] -- cgit v1.2.3