Michilis/lnbits
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: plan

Browse Source
This commit is contained in:
Vlad Stan
2026-05-13 15:23:29 +03:00
parent b8055b4f51
commit 272cb9bcc9
2 changed files with 230 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lnbits/fiat/square.py
+127 -25
View File
@@ -52,6 +52,11 @@ class SquareCreateInvoiceOptions(BaseModel):
subscription: SquareSubscriptionOptions | None = None
class SquareSubscriptionCheckoutInfo(BaseModel):
plan_variation_id: str
price_money: dict[str, Any]
class SquareWallet(FiatProvider):
"""https://developer.squareup.com/reference/square"""
@@ -199,23 +204,20 @@ class SquareWallet(FiatProvider):
payment_options.extra["subscription_request_id"] = (
payment_options.subscription_request_id
)
print("### create_subscription", subscription_id, quantity, payment_options)
try:
price_money = await self._get_subscription_price_money(subscription_id)
print("### price_money", price_money)
checkout_info = await self._get_subscription_checkout_info(subscription_id)
metadata = self._serialize_metadata(payment_options)
print("### metadata", metadata)
payload = {
"idempotency_key": payment_options.subscription_request_id,
"description": metadata,
"quick_pay": {
"name": (payment_options.memo or "LNbits Subscription")[:255],
"price_money": price_money,
"price_money": checkout_info.price_money,
"location_id": self.location_id,
},
"checkout_options": {
"redirect_url": success_url,
"subscription_plan_id": subscription_id,
"subscription_plan_id": checkout_info.plan_variation_id,
},
"payment_note": metadata,
}
@@ -224,7 +226,6 @@ class SquareWallet(FiatProvider):
)
r.raise_for_status()
data = r.json()
print("### response data", data)
payment_link = data.get("payment_link") or {}
url = payment_link.get("url")
if not url:
@@ -338,33 +339,134 @@ class SquareWallet(FiatProvider):
r.raise_for_status()
return r.json().get("payment") or {}
async def _get_subscription_price_money(
self, plan_variation_id: str
) -> dict[str, Any]:
r = await self.client.get(f"/v2/catalog/object/{plan_variation_id}")
r.raise_for_status()
catalog_object = r.json().get("object") or {}
if catalog_object.get("type") != "SUBSCRIPTION_PLAN_VARIATION":
raise ValueError("Square subscription ID must be a plan variation ID.")
async def _get_subscription_checkout_info(
self, subscription_plan_id: str
) -> SquareSubscriptionCheckoutInfo:
catalog_object = await self._get_catalog_object(subscription_plan_id)
if catalog_object.get("type") == "SUBSCRIPTION_PLAN":
return await self._get_plan_checkout_info(catalog_object)
variation_data = catalog_object.get("subscription_plan_variation_data") or {}
if catalog_object.get("type") == "SUBSCRIPTION_PLAN_VARIATION":
price_money = await self._get_subscription_price_money(
catalog_object,
)
plan_variation_id = catalog_object.get("id")
if not plan_variation_id:
raise ValueError("Square subscription plan variation is missing an ID.")
return SquareSubscriptionCheckoutInfo(
plan_variation_id=plan_variation_id,
price_money=price_money,
)
raise ValueError(
"Square subscription ID must be a plan ID or plan variation ID."
)
async def _get_plan_checkout_info(
self, catalog_object: dict[str, Any]
) -> SquareSubscriptionCheckoutInfo:
plan_data = catalog_object.get("subscription_plan_data") or {}
plan_variations = plan_data.get("subscription_plan_variations") or []
plan_variation = next(
(
variation
for variation in plan_variations
if not variation.get("is_deleted")
),
None,
)
if not plan_variation:
raise ValueError("Square subscription plan is missing a variation.")
price_money = await self._get_subscription_price_money(
plan_variation,
eligible_item_ids=plan_data.get("eligible_item_ids") or [],
)
plan_variation_id = plan_variation.get("id")
if not plan_variation_id:
raise ValueError("Square subscription plan variation is missing an ID.")
return SquareSubscriptionCheckoutInfo(
plan_variation_id=plan_variation_id,
price_money=price_money,
)
async def _get_catalog_object(self, object_id: str) -> dict[str, Any]:
r = await self.client.get(f"/v2/catalog/object/{object_id}")
r.raise_for_status()
return r.json().get("object") or {}
async def _get_subscription_price_money(
self,
plan_variation: dict[str, Any],
eligible_item_ids: list[str] | None = None,
) -> dict[str, Any]:
variation_data = plan_variation.get("subscription_plan_variation_data") or {}
phases = variation_data.get("phases") or []
for phase in phases:
pricing = phase.get("pricing") or {}
price_money = pricing.get("price_money") or phase.get(
"recurring_price_money"
)
if (
price_money
and price_money.get("amount") is not None
and price_money.get("currency")
):
return {
"amount": int(price_money["amount"]),
"currency": price_money["currency"].upper(),
}
parsed_price_money = self._parse_price_money(price_money)
if parsed_price_money:
return parsed_price_money
if pricing.get("type") == "RELATIVE":
return await self._get_relative_subscription_price_money(
eligible_item_ids or []
)
raise ValueError("Square subscription plan variation is missing price_money.")
async def _get_relative_subscription_price_money(
self, eligible_item_ids: list[str]
) -> dict[str, Any]:
if len(eligible_item_ids) != 1:
raise ValueError(
"Square relative subscription plan must have exactly one item."
)
item = await self._get_catalog_object(eligible_item_ids[0])
item_variations: list[dict[str, Any]] = []
if item.get("type") == "ITEM":
item_variations = (item.get("item_data") or {}).get("variations") or []
elif item.get("type") == "ITEM_VARIATION":
item_variations = [item]
item_variation = next(
(
variation
for variation in item_variations
if not variation.get("is_deleted")
),
None,
)
if not item_variation:
raise ValueError("Square subscription item is missing a variation.")
price_money = self._parse_price_money(
(item_variation.get("item_variation_data") or {}).get("price_money")
)
if price_money:
return price_money
raise ValueError("Square subscription item variation is missing price_money.")
def _parse_price_money(
self, price_money: dict[str, Any] | None
) -> dict[str, Any] | None:
if (
price_money
and price_money.get("amount") is not None
and price_money.get("currency")
):
return {
"amount": int(price_money["amount"]),
"currency": price_money["currency"].upper(),
}
return None
def _status_from_payment(self, payment: dict[str, Any]) -> FiatPaymentStatus:
status = (payment.get("status") or "").upper()
if status == "COMPLETED":
tests/unit/test_fiat_providers.py
+103
View File
@@ -422,6 +422,7 @@ async def test_square_wallet_create_subscription(settings: Settings):
json_data={
"object": {
"type": "SUBSCRIPTION_PLAN_VARIATION",
"id": "PLAN_VARIATION_123",
"subscription_plan_variation_data": {
"phases": [
{
@@ -481,6 +482,108 @@ async def test_square_wallet_create_subscription(settings: Settings):
assert metadata[3:] == ["link-1", "Monthly Gold"]
@pytest.mark.anyio
async def test_square_wallet_create_subscription_from_plan_id(settings: Settings):
settings.square_api_endpoint = "https://connect.squareupsandbox.com"
settings.square_access_token = "square-token"
settings.square_location_id = "LOC123"
settings.square_api_version = "2026-01-22"
settings.square_payment_success_url = "https://lnbits.example/success"
wallet = SquareWallet()
client = MockHTTPClient(
[
MockHTTPResponse(
json_data={
"object": {
"type": "SUBSCRIPTION_PLAN",
"id": "PLAN123",
"subscription_plan_data": {
"name": "LNbits Test Weekly Personal Plan",
"subscription_plan_variations": [
{
"type": "SUBSCRIPTION_PLAN_VARIATION",
"id": "PLAN_VARIATION_123",
"subscription_plan_variation_data": {
"name": "LNbits Test Weekly Personal Plan",
"phases": [
{
"uid": "PHASE123",
"cadence": "WEEKLY",
"ordinal": 0,
"pricing": {"type": "RELATIVE"},
}
],
"subscription_plan_id": "PLAN123",
},
}
],
"eligible_item_ids": ["ITEM123"],
"all_items": False,
},
}
}
),
MockHTTPResponse(
json_data={
"object": {
"type": "ITEM",
"id": "ITEM123",
"item_data": {
"name": "LNbits Test Weekly Personal Plan",
"variations": [
{
"type": "ITEM_VARIATION",
"id": "ITEM_VARIATION_123",
"item_variation_data": {
"item_id": "ITEM123",
"name": "Regular",
"pricing_type": "FIXED_PRICING",
"price_money": {
"amount": 1500,
"currency": "USD",
},
},
}
],
},
}
}
),
MockHTTPResponse(
json_data={
"payment_link": {
"id": "plink_123",
"url": "https://square.link/u/sub_123",
}
}
),
]
)
wallet.client = client # type: ignore[assignment]
response = await wallet.create_subscription(
"PLAN123",
1,
FiatSubscriptionPaymentOptions(
wallet_id="wallet_1",
memo="Weekly Plan",
success_url="https://lnbits.example/success",
),
)
assert response.ok is True
assert client.calls[0][0] == "/v2/catalog/object/PLAN123"
assert client.calls[1][0] == "/v2/catalog/object/ITEM123"
assert client.calls[2][0] == "/v2/online-checkout/payment-links"
payload = client.calls[2][1]["json"]
assert payload["quick_pay"]["price_money"] == {"amount": 1500, "currency": "USD"}
assert payload["checkout_options"] == {
"redirect_url": "https://lnbits.example/success",
"subscription_plan_id": "PLAN_VARIATION_123",
}
@pytest.mark.anyio
async def test_square_wallet_create_subscription_invoice(settings: Settings):
settings.square_api_endpoint = "https://connect.squareupsandbox.com"