mindroot.coreplugins.subscriptions package

Submodules

mindroot.coreplugins.subscriptions.credit_integration module

class mindroot.coreplugins.subscriptions.credit_integration.CreditIntegration

Bases: object

Handles integration with the credits plugin for subscription-related credit operations

async allocate_subscription_credits(username: str, amount: float, subscription_id: str, metadata: Dict[str, Any]) → float

Allocate credits to a user based on their subscription

Parameters:

• username – User to allocate credits to

• amount – Amount of credits to allocate

• subscription_id – Subscription ID for reference

• metadata – Additional information about the allocation

Returns:

New credit balance

Return type:

float

async check_credit_balance(username: str) → float

Get current credit balance for a user

mindroot.coreplugins.subscriptions.default_templates module

mindroot.coreplugins.subscriptions.mod module

class mindroot.coreplugins.subscriptions.mod.SubscriptionsPlugin(base_path: str)

Bases: object

create_components()

Create fresh instances of all subscription system components

async mindroot.coreplugins.subscriptions.mod.activate_plan_feature(feature_id: str, context=None) → Dict[str, Any]

Activate a plan feature

Parameters:

• feature_id – ID of the feature to activate

• context – Request context

Returns:

Status result

Return type:

Dict

async mindroot.coreplugins.subscriptions.mod.activate_subscription_plan(plan_id: str, context=None) → Dict[str, Any]

Activate a subscription plan

Parameters:

• plan_id – ID of the plan to activate

• context – Request context

Returns:

Status result

Return type:

Dict

async mindroot.coreplugins.subscriptions.mod.cancel_subscription(subscription_id: str, at_period_end: bool = True, context=None) → Dict[str, Any]

Cancel a user subscription

Parameters:

• subscription_id – ID of the subscription to cancel

• at_period_end – Whether to cancel at period end or immediately

• context – Request context

Returns:

Updated subscription details

Return type:

Dict

async mindroot.coreplugins.subscriptions.mod.create_page_template(template_data: Dict[str, Any], context=None) → Dict[str, Any]

Create a new page template

Parameters:

• template_data – Template details

• context – Request context

Returns:

Created template details

Return type:

Dict

async mindroot.coreplugins.subscriptions.mod.create_plan_feature(feature_data: Dict[str, Any], context=None) → Dict[str, Any]

Create a new plan feature

Parameters:

• feature_data – Feature details

• context – Request context

Returns:

Created feature details

Return type:

Dict

async mindroot.coreplugins.subscriptions.mod.create_subscription_plan(plan_data: Dict[str, Any], context=None) → Dict[str, Any]

Create a new subscription plan

Parameters:

• plan_data – Plan details including name, description, price, etc.

• context – Request context

Returns:

Created plan details

Return type:

Dict

async mindroot.coreplugins.subscriptions.mod.deactivate_plan_feature(feature_id: str, context=None) → Dict[str, Any]

Deactivate a plan feature

Parameters:

• feature_id – ID of the feature to deactivate

• context – Request context

Returns:

Status result

Return type:

Dict

async mindroot.coreplugins.subscriptions.mod.deactivate_subscription_plan(plan_id: str, context=None) → Dict[str, Any]

Deactivate a subscription plan

Parameters:

• plan_id – ID of the plan to deactivate

• context – Request context

Returns:

Status result

Return type:

Dict

async mindroot.coreplugins.subscriptions.mod.get_available_features(active_only: bool = True, context=None) → List[Dict[str, Any]]

Get all available plan features

Parameters:

• active_only – Whether to return only active features

• context – Request context

Returns:

List of feature details

Return type:

List[Dict]

mindroot.coreplugins.subscriptions.mod.get_base_path(context) → str

Get the base path for subscription data storage

async mindroot.coreplugins.subscriptions.mod.get_my_subscriptions(params, context=None)

Get current user’s subscriptions

Example

{ “get_my_subscriptions”: {} }

async mindroot.coreplugins.subscriptions.mod.get_page_templates(context=None) → List[Dict[str, Any]]

Get all page templates

Parameters:

context – Request context

Returns:

List of template details

Return type:

List[Dict]

async mindroot.coreplugins.subscriptions.mod.get_subscription_by_provider_id(provider_subscription_id: str, context=None) → Dict | None

Get subscription by provider subscription ID (e.g., Stripe subscription ID)

Parameters:

• provider_subscription_id – Provider’s subscription ID

• context – Request context

Returns:

Subscription details or None if not found

Return type:

Optional[Dict]

async mindroot.coreplugins.subscriptions.mod.get_subscription_plan(plan_id: str, context=None) → Dict[str, Any] | None

Get a subscription plan by ID

Parameters:

• plan_id – ID of the plan to retrieve

• context – Request context

Returns:

Plan details or None if not found

Return type:

Optional[Dict]

async mindroot.coreplugins.subscriptions.mod.get_subscription_plans(active_only: bool = True, context=None) → List[Dict[str, Any]]

Get all subscription plans

Parameters:

• active_only – Whether to return only active plans

• context – Request context

Returns:

List of plan details

Return type:

List[Dict]

async mindroot.coreplugins.subscriptions.mod.get_user_subscriptions(username: str, context=None) → List[Dict[str, Any]]

Get all subscriptions for a user

Parameters:

• username – User to get subscriptions for

• context – Request context

Returns:

List of subscription details

Return type:

List[Dict]

async mindroot.coreplugins.subscriptions.mod.initialize_default_features(storage)

Initialize default plan features

async mindroot.coreplugins.subscriptions.mod.initialize_default_template(storage)

Initialize default page template

async mindroot.coreplugins.subscriptions.mod.list_subscription_plans(params, context=None)

List available subscription plans

Example

{ “list_subscription_plans”: {} }

async mindroot.coreplugins.subscriptions.mod.process_subscription_event(event_data: Dict[str, Any], context=None) → Dict[str, Any]

Process normalized subscription events from any payment provider

Parameters:

• event_data – Normalized event data with provider, event type, etc.

• context – Request context

Returns:

Processing result

Return type:

Dict

async mindroot.coreplugins.subscriptions.mod.set_default_template(template_id: str, context=None) → Dict[str, Any]

Set a template as the default

Parameters:

• template_id – ID of the template to set as default

• context – Request context

Returns:

Status result

Return type:

Dict

async mindroot.coreplugins.subscriptions.mod.startup(app, context=None)

Startup tasks

async mindroot.coreplugins.subscriptions.mod.update_page_template(template_id: str, template_data: Dict[str, Any], context=None) → Dict[str, Any]

Update an existing page template

Parameters:

• template_id – ID of the template to update

• template_data – Updated template details

• context – Request context

Returns:

Updated template details

Return type:

Dict

async mindroot.coreplugins.subscriptions.mod.update_plan_feature(feature_id: str, feature_data: Dict[str, Any], context=None) → Dict[str, Any]

Update an existing plan feature

Parameters:

• feature_id – ID of the feature to update

• feature_data – Updated feature details

• context – Request context

Returns:

Updated feature details

Return type:

Dict

async mindroot.coreplugins.subscriptions.mod.update_subscription_plan(plan_id: str, plan_data: Dict[str, Any], context=None) → Dict[str, Any]

Update an existing subscription plan

Parameters:

• plan_id – ID of the plan to update

• plan_data – Updated plan details

• context – Request context

Returns:

Updated plan details

Return type:

Dict

async mindroot.coreplugins.subscriptions.mod.update_subscription_status(subscription_id: str, status: str, context=None) → Dict | None

Update a subscription’s status

Parameters:

• subscription_id – Internal subscription ID

• status – New status (active, canceled, etc.)

• context – Request context

Returns:

Updated subscription details or None if not found

Return type:

Optional[Dict]

mindroot.coreplugins.subscriptions.models module

class mindroot.coreplugins.subscriptions.models.PageTemplate(template_id: str, name: str, description: str, html_template: str, css_template: str, js_template: str, is_default: bool = False)

Bases: object

css_template: str

description: str

classmethod from_dict(data: dict) → PageTemplate

html_template: str

is_default: bool = False

js_template: str

name: str

template_id: str

to_dict() → dict

class mindroot.coreplugins.subscriptions.models.PlanFeature(feature_id: str, name: str, description: str, type: str, display_order: int = 0, options: List[str] = <factory>, default_value: Any = None, active: bool = True)

Bases: object

active: bool = True

default_value: Any = None

description: str

display_order: int = 0

feature_id: str

classmethod from_dict(data: dict) → PlanFeature

name: str

options: List[str]

to_dict() → dict

type: str

class mindroot.coreplugins.subscriptions.models.SubscriptionPlan(plan_id: str, name: str, description: str, price: float, credits_per_cycle: float, currency: str = 'USD', interval: str = 'month', features: Dict[str, Any] = <factory>, metadata: Dict[str, Any] = <factory>, active: bool = True)

Bases: object

active: bool = True

credits_per_cycle: float

currency: str = 'USD'

description: str

features: Dict[str, Any]

classmethod from_dict(data: dict) → SubscriptionPlan

interval: str = 'month'

metadata: Dict[str, Any]

name: str

plan_id: str

price: float

to_dict() → dict

class mindroot.coreplugins.subscriptions.models.UserSubscription(subscription_id: str, username: str, plan_id: str, status: str, current_period_start: datetime.datetime, current_period_end: datetime.datetime, cancel_at_period_end: bool = False, payment_provider: str = 'stripe', provider_subscription_id: Optional[str] = None, metadata: Dict[str, Any] = <factory>, created_at: datetime.datetime = <factory>, updated_at: datetime.datetime = <factory>)

Bases: object

cancel_at_period_end: bool = False

created_at: datetime

current_period_end: datetime

current_period_start: datetime

classmethod from_dict(data: dict) → UserSubscription

metadata: Dict[str, Any]

payment_provider: str = 'stripe'

plan_id: str

provider_subscription_id: str | None = None

status: str

subscription_id: str

to_dict() → dict

updated_at: datetime

username: str

mindroot.coreplugins.subscriptions.router module

async mindroot.coreplugins.subscriptions.router.api_activate_feature(feature_id: str, request: Request)

Activate a plan feature

async mindroot.coreplugins.subscriptions.router.api_activate_plan(plan_id: str, request: Request)

Activate a subscription plan

async mindroot.coreplugins.subscriptions.router.api_cancel_subscription(subscription_id: str, request: Request, at_period_end: bool = True)

Cancel a subscription

async mindroot.coreplugins.subscriptions.router.api_create_checkout(plan_id: str, request: Request, provider: str = Query(stripe))

Create checkout session for subscription

async mindroot.coreplugins.subscriptions.router.api_create_feature(request: Request)

Create a new plan feature

async mindroot.coreplugins.subscriptions.router.api_create_plan(request: Request)

Create a new subscription plan

async mindroot.coreplugins.subscriptions.router.api_create_template(request: Request)

Create a new page template

async mindroot.coreplugins.subscriptions.router.api_deactivate_feature(feature_id: str, request: Request)

Deactivate a plan feature

async mindroot.coreplugins.subscriptions.router.api_deactivate_plan(plan_id: str, request: Request)

Deactivate a subscription plan

async mindroot.coreplugins.subscriptions.router.api_get_all_plans(request: Request, active_only: bool = True)

Get all subscription plans

async mindroot.coreplugins.subscriptions.router.api_get_features(request: Request, active_only: bool = False)

Get all plan features

async mindroot.coreplugins.subscriptions.router.api_get_my_subscriptions(request: Request)

Get current user’s subscriptions

async mindroot.coreplugins.subscriptions.router.api_get_plans(request: Request)

Get available subscription plans

async mindroot.coreplugins.subscriptions.router.api_get_templates(request: Request)

Get all page templates

async mindroot.coreplugins.subscriptions.router.api_set_default_template(template_id: str, request: Request)

Set a template as the default

async mindroot.coreplugins.subscriptions.router.api_update_feature(feature_id: str, request: Request)

Update an existing plan feature

async mindroot.coreplugins.subscriptions.router.api_update_plan(plan_id: str, request: Request)

Update an existing subscription plan

async mindroot.coreplugins.subscriptions.router.api_update_template(template_id: str, request: Request)

Update an existing page template

async mindroot.coreplugins.subscriptions.router.subscription_page(request: Request)

Public subscription page

mindroot.coreplugins.subscriptions.storage module

class mindroot.coreplugins.subscriptions.storage.SubscriptionStorage(base_path: str)

Bases: object

async get_all_features(active_only: bool = True) → List[PlanFeature]

Get all plan features

async get_all_plans(active_only: bool = True) → List[SubscriptionPlan]

Get all subscription plans

async get_all_templates() → List[PageTemplate]

Get all page templates

async get_default_template() → PageTemplate | None

Get the default page template

async get_feature(feature_id: str) → PlanFeature | None

Get a plan feature by ID

async get_plan(plan_id: str) → SubscriptionPlan | None

Get a subscription plan by ID

async get_subscription(subscription_id: str) → UserSubscription | None

Get a subscription by ID

async get_subscriptions_by_provider_id(provider: str, provider_subscription_id: str) → List[UserSubscription]

Get subscriptions by provider subscription ID

async get_template(template_id: str) → PageTemplate | None

Get a page template by ID

async get_user_subscriptions(username: str) → List[UserSubscription]

Get all subscriptions for a user

async set_default_template(template_id: str) → bool

Set a template as the default

async store_feature(feature: PlanFeature) → None

Store a plan feature

async store_plan(plan: SubscriptionPlan) → None

Store a subscription plan

async store_subscription(subscription: UserSubscription) → None

Store a user subscription

async store_template(template: PageTemplate) → None

Store a page template

async update_feature(feature: PlanFeature) → None

Update a plan feature

async update_plan(plan: SubscriptionPlan) → None

Update a subscription plan

async update_subscription(subscription: UserSubscription) → None

Update a user subscription

mindroot.coreplugins.subscriptions.stripe_integration module

class mindroot.coreplugins.subscriptions.stripe_integration.StripeIntegration

Bases: object

Handles integration with the Stripe payment plugin for subscription operations

async cancel_subscription(provider_subscription_id: str, at_period_end: bool = True) → bool

Cancel a Stripe subscription

Parameters:

• provider_subscription_id – Stripe subscription ID

• at_period_end – Whether to cancel at period end or immediately

Returns:

Success status

Return type:

bool

async create_checkout_session(username: str, plan: SubscriptionPlan) → str

Create a Stripe checkout session for subscription

Parameters:

• username – User to create checkout for

• plan – Subscription plan details

Returns:

Checkout URL

Return type:

str

mindroot.coreplugins.subscriptions.subscription_manager module

class mindroot.coreplugins.subscriptions.subscription_manager.SubscriptionManager(storage: SubscriptionStorage, base_path: str)

Bases: object

async cancel_subscription(subscription_id: str, at_period_end: bool = True) → UserSubscription

Cancel a user subscription

async create_plan(plan_data: Dict[str, Any]) → SubscriptionPlan

Create a new subscription plan

async create_subscription(username: str, plan_id: str, provider_data: Dict[str, Any]) → UserSubscription

Create a new user subscription

async deactivate_plan(plan_id: str) → None

Deactivate a subscription plan

async get_plan(plan_id: str) → SubscriptionPlan | None

Get a subscription plan by ID

async get_subscriptions_by_provider_id(provider: str, provider_subscription_id: str) → List[UserSubscription]

Get subscriptions by provider subscription ID

async get_user_subscriptions(username: str) → List[UserSubscription]

Get all subscriptions for a user

async list_plans(active_only: bool = True) → List[SubscriptionPlan]

List all subscription plans

async update_plan(plan_id: str, plan_data: Dict[str, Any]) → SubscriptionPlan

Update an existing subscription plan

async update_subscription_period(subscription_id: str, period_start: datetime, period_end: datetime) → UserSubscription

Update a subscription’s billing period

async update_subscription_status(subscription_id: str, status: str) → UserSubscription

Update a subscription’s status

mindroot.coreplugins.subscriptions.webhook_handler module

class mindroot.coreplugins.subscriptions.webhook_handler.WebhookHandler(subscription_manager: SubscriptionManager, credit_integration: CreditIntegration)

Bases: object

Handles webhook events from payment providers for subscription events

async handle_stripe_event(event: Dict[str, Any]) → Dict | None

Process Stripe webhook events related to subscriptions

Parameters:

event – Stripe event object

Returns:

Processing result

Return type:

Optional[Dict]

Module contents