Skip to content
Back to top
View this page

# 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: str | None = 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