feat: add ServerItem, AsyncServerItem, and EmailDeliveryInfo models

- ServerItem wraps Item with increase/decrease/tryDecrease quantity methods
- AsyncServerItem provides async counterpart for quantity operations
- EmailDeliveryInfo model for delivery statistics (delivered/bounced/complained/total)
- Export new models from models package
This commit is contained in:
Ejiro Asiuwhu
2026-03-25 01:47:35 +01:00
parent e8ce41bfbb
commit 0e6afdc9a7
3 changed files with 153 additions and 1 deletions
@@ -10,7 +10,8 @@ from stack_auth.models.api_keys import (
from stack_auth.models.contact_channels import ContactChannel
from stack_auth.models.notifications import NotificationCategory
from stack_auth.models.oauth import OAuthConnection, OAuthProvider
from stack_auth.models.payments import Item, Product
from stack_auth.models.email import EmailDeliveryInfo
from stack_auth.models.payments import AsyncServerItem, Item, Product, ServerItem
from stack_auth.models.permissions import ProjectPermission, TeamPermission
from stack_auth.models.projects import OAuthProviderConfig, Project, ProjectConfig
from stack_auth.models.sessions import ActiveSession, GeoInfo
@@ -41,5 +42,8 @@ __all__ = [
"OAuthProvider",
"Product",
"Item",
"ServerItem",
"AsyncServerItem",
"EmailDeliveryInfo",
"NotificationCategory",
]
@@ -0,0 +1,14 @@
"""Email models for Stack Auth."""
from __future__ import annotations
from stack_auth.models._base import StackAuthModel
class EmailDeliveryInfo(StackAuthModel):
"""Email delivery statistics."""
delivered: int = 0
bounced: int = 0
complained: int = 0
total: int = 0
@@ -2,6 +2,8 @@
from __future__ import annotations
from typing import Any
from pydantic import Field
from stack_auth.models._base import StackAuthModel
@@ -25,3 +27,135 @@ class Product(StackAuthModel):
is_server_only: bool = Field(False, alias="isServerOnly")
stackable: bool = False
type: str = "one_time"
class ServerItem:
"""Server-side item with methods to modify quantity.
Wraps an :class:`Item` and provides ``increase_quantity``,
``decrease_quantity``, and ``try_decrease_quantity`` methods that
communicate with the Stack Auth API via the HTTP client.
"""
def __init__(
self,
item: Item,
*,
_client: Any,
_customer_path: str,
_item_id: str,
_customer_id_field: str,
_customer_id_value: str,
) -> None:
self.display_name = item.display_name
self.quantity = item.quantity
self.non_negative_quantity = item.non_negative_quantity
self._client = _client
self._customer_path = _customer_path
self._item_id = _item_id
self._customer_id_field = _customer_id_field
self._customer_id_value = _customer_id_value
def _quantity_body(self, quantity: int) -> dict[str, Any]:
return {
self._customer_id_field: self._customer_id_value,
"item_id": self._item_id,
"quantity": quantity,
}
def increase_quantity(self, amount: int) -> None:
"""Increase this item's quantity by *amount*."""
self._client.request(
"POST",
"/internal/items/quantity-changes",
body=self._quantity_body(amount),
)
def decrease_quantity(self, amount: int) -> None:
"""Decrease this item's quantity by *amount*."""
self._client.request(
"POST",
"/internal/items/quantity-changes",
body=self._quantity_body(-amount),
)
def try_decrease_quantity(self, amount: int) -> bool:
"""Try to decrease this item's quantity by *amount*.
Returns ``True`` if the decrease succeeded, ``False`` otherwise.
"""
data = self._client.request(
"POST",
"/internal/items/try-decrease",
body={
self._customer_id_field: self._customer_id_value,
"item_id": self._item_id,
"amount": amount,
},
)
return bool(data.get("success", False)) if data else False
class AsyncServerItem:
"""Async server-side item with methods to modify quantity.
Async counterpart of :class:`ServerItem`.
"""
def __init__(
self,
item: Item,
*,
_client: Any,
_customer_path: str,
_item_id: str,
_customer_id_field: str,
_customer_id_value: str,
) -> None:
self.display_name = item.display_name
self.quantity = item.quantity
self.non_negative_quantity = item.non_negative_quantity
self._client = _client
self._customer_path = _customer_path
self._item_id = _item_id
self._customer_id_field = _customer_id_field
self._customer_id_value = _customer_id_value
def _quantity_body(self, quantity: int) -> dict[str, Any]:
return {
self._customer_id_field: self._customer_id_value,
"item_id": self._item_id,
"quantity": quantity,
}
async def increase_quantity(self, amount: int) -> None:
"""Increase this item's quantity by *amount*."""
await self._client.request(
"POST",
"/internal/items/quantity-changes",
body=self._quantity_body(amount),
)
async def decrease_quantity(self, amount: int) -> None:
"""Decrease this item's quantity by *amount*."""
await self._client.request(
"POST",
"/internal/items/quantity-changes",
body=self._quantity_body(-amount),
)
async def try_decrease_quantity(self, amount: int) -> bool:
"""Try to decrease this item's quantity by *amount*.
Returns ``True`` if the decrease succeeded, ``False`` otherwise.
"""
data = await self._client.request(
"POST",
"/internal/items/try-decrease",
body={
self._customer_id_field: self._customer_id_value,
"item_id": self._item_id,
"amount": amount,
},
)
return bool(data.get("success", False)) if data else False