# Copyright (c) 2025 Ferrosoft GmbH. All rights reserved.
from abc import ABC, abstractmethod
from django.contrib.auth.mixins import AccessMixin
from django.core.exceptions import ImproperlyConfigured
from django.http import HttpRequest
from django.shortcuts import redirect
from django.urls import reverse
from django.utils.translation import gettext_lazy as _
from ferrosoft.apps.ferrobase.auth.roles import Permissions, Verb
from ferrosoft.apps.ferrobase.filter import (
OrganizationFilterSet,
SubscriptionPlanFilterSet,
)
from ferrosoft.apps.ferrobase.forms import (
OrganizationForm,
SubscriptionPlanForm,
FeatureAssignmentForm,
)
from ferrosoft.apps.ferrobase.forms.copy import copy_form_factory, copy_fields
from ferrosoft.apps.ferrobase.models import (
Organization,
User,
SubscriptionPlan,
FeatureAssignment,
)
from ferrosoft.apps.ferrobase.tables.models import (
OrganizationTable,
TenantTable,
SubscriptionPlanTable,
FeatureAssignmentTable,
)
from ferrosoft.apps.ferrobase.views.tenant import (
TenantCreateView,
)
from ferrosoft.apps.ferrobase.views.generic import (
TableView,
TemplateView,
DependentCreateView,
CreateView,
UpdateView,
DeleteView,
model_crud_routes,
CopyView,
)
from ferrosoft.config import require_settings
[docs]
class FeatureLimiter(ABC):
"""Get current count of feature to determine whether the limit is reached."""
[docs]
@abstractmethod
def get_count(self, request: HttpRequest) -> int:
"""Return the current count of the limited feature for ``request``."""
[docs]
class FeatureRequiredMixin(AccessMixin):
"""Mixin gating a view behind a subscription-plan feature and optional quota.
Anonymous users bypass the check (they fall through to the normal auth
flow). Authenticated users must either be on a plan with
``enable_all_features`` or hold an explicit :class:`FeatureAssignment`
for :attr:`feature_required`. When a :attr:`feature_limiter` is set,
its count is compared against ``assignment.limit``.
"""
feature_required: str | None = None
feature_limiter: FeatureLimiter | None = None
[docs]
def handle_no_permission(self):
"""Redirect authenticated users to the upgrade page; defer otherwise."""
if self.request.user.is_anonymous:
# Anonymous users should never be redirected to upgrade page.
return super().handle_no_permission()
return redirect(reverse("ferrobase:upgrade_plan"))
[docs]
def get_feature_required(self):
"""Return :attr:`feature_required` or raise :class:`ImproperlyConfigured`."""
if self.feature_required is None:
raise ImproperlyConfigured(
f"{self.__class__.__name__} is missing the "
f"feature_required attribute. Define "
f"{self.__class__.__name__}.feature_required, or override "
f"{self.__class__.__name__}.get_feature_required()."
)
return self.feature_required
[docs]
def validate_feature(
self, request: HttpRequest
) -> tuple[FeatureAssignment | None, SubscriptionPlan | None]:
"""Look up the active assignment and plan for the required feature."""
# noinspection PyUnresolvedReferences
return request.user.get_assigned_feature(self.get_feature_required())
[docs]
def limit_reached(
self, request: HttpRequest, assignment: FeatureAssignment
) -> bool:
"""Return ``True`` if the configured limiter's count meets ``assignment.limit``."""
if self.feature_limiter is None:
return False
return self.feature_limiter.get_count(request) >= assignment.limit
[docs]
def dispatch(self, request, *args, **kwargs):
"""Enforce the feature/limit policy before dispatching to the view."""
if request.user.is_anonymous:
# Anonymous users don't have a subscription plan.
return super().dispatch(request, *args, **kwargs)
assignment, plan = self.validate_feature(request)
assignment_required = plan is not None and not plan.enable_all_features
# fmt: off
if assignment_required and (assignment is None or self.limit_reached(request, assignment)):
return self.handle_no_permission()
# fmt: on
# noinspection PyUnresolvedReferences
return super().dispatch(request, *args, **kwargs)
[docs]
class UpgradePlanView(TemplateView):
"""Static page shown when a feature requires a plan upgrade."""
activity_title = _("Upgrade Required")
template_name = "ferrobase/subscriptionplan/upgrade.html"
[docs]
def get_context_data(self, **kwargs):
"""Expose the marketing plan URL and sales contact from settings."""
settings = require_settings(
"FERROBASE_SUBSCRIPTION_PLAN_URL",
"FERROBASE_SALES_ADDRESS",
)
return super().get_context_data(**kwargs) | {
"plan_url": settings["FERROBASE_SUBSCRIPTION_PLAN_URL"],
"sales_address": settings["FERROBASE_SALES_ADDRESS"],
}
[docs]
class FeatureAssignmentCreateView(DependentCreateView):
"""Create a feature assignment under an existing :class:`SubscriptionPlan`."""
form_class = FeatureAssignmentForm
model = FeatureAssignment
parent_model = SubscriptionPlan
[docs]
class FeatureAssignmentUpdateView(UpdateView):
"""Update a feature assignment; redirects back to its plan on success."""
form_class = FeatureAssignmentForm
model = FeatureAssignment
[docs]
def get_success_url(self):
"""Return to the parent plan's update page."""
return self.object.plan.get_absolute_url()
[docs]
class FeatureAssignmentDeleteView(DeleteView):
"""Delete a feature assignment; redirects back to its plan."""
model = FeatureAssignment
[docs]
def get_success_url(self):
"""Return to the parent plan's update page."""
return self.object.plan.get_absolute_url()
feature_assignment_urls = model_crud_routes(
FeatureAssignment,
create_view=FeatureAssignmentCreateView,
update_view=FeatureAssignmentUpdateView,
delete_view=FeatureAssignmentDeleteView,
)
[docs]
class OrganizationTableView(TableView):
"""Filterable table of organisations."""
filterset_class = OrganizationFilterSet
model = Organization
ordering = "name"
table_class = OrganizationTable
[docs]
class OrganizationCreateView(CreateView):
"""Create an organisation."""
form_class = OrganizationForm
model = Organization
[docs]
class OrganizationUpdateView(UpdateView):
"""Update an organisation and display its tenants."""
form_class = OrganizationForm
model = Organization
template_name = "ferrobase/organization/update.html"
[docs]
def get_context_data(self, **kwargs):
"""Add the per-organisation tenant table to the template context."""
return super().get_context_data(**kwargs) | {
"tenant_table": TenantTable(
self.object.tenants.order_by("name"),
create_url=reverse(
"ferrobase:tenant_create",
query={
TenantCreateView.parent_query: str(self.object.pk),
},
),
),
}
[docs]
class OrganizationDeleteView(DeleteView):
"""Delete an organisation."""
model = Organization
organization_urls = model_crud_routes(
Organization,
create_view=OrganizationCreateView,
read_view=OrganizationTableView,
update_view=OrganizationUpdateView,
delete_view=OrganizationDeleteView,
)
[docs]
class SubscriptionPlanTableView(TableView):
"""Filterable table of subscription plans."""
filterset_class = SubscriptionPlanFilterSet
model = SubscriptionPlan
ordering = "name"
table_class = SubscriptionPlanTable
[docs]
class SubscriptionPlanCopyView(CopyView):
"""Create a new subscription plan seeded from an existing one's settings."""
[docs]
@staticmethod
def prepare_copy(source: SubscriptionPlan, target: SubscriptionPlan, **kwargs):
"""Return preparators that clone every :class:`FeatureAssignment` of ``source``.
The cloning is deferred via a returned callable so the copy form
factory can invoke it after the new plan has been persisted.
"""
def copy_features():
features = [
copy_fields(
source_feature,
FeatureAssignment(),
fields=["feature", "limit"],
plan=target,
)
for source_feature in FeatureAssignment.objects.filter(plan=source)
]
if len(features) == 0:
return
FeatureAssignment.objects.bulk_create(features)
return [copy_features]
# noinspection PyTypeChecker
form_class = copy_form_factory(
SubscriptionPlan,
fields=["name"],
fields_to_copy=["max_tonnage", "enable_all_features"],
preparator=prepare_copy,
)
model = SubscriptionPlan
[docs]
class SubscriptionPlanCreateView(CreateView):
"""Create a subscription plan."""
form_class = SubscriptionPlanForm
model = SubscriptionPlan
[docs]
class SubscriptionPlanUpdateView(UpdateView):
"""Update a plan and display its associated feature assignments."""
add_copy_action = True
form_class = SubscriptionPlanForm
model = SubscriptionPlan
template_name = "ferrobase/subscriptionplan/update.html"
[docs]
def get_context_data(self, **kwargs):
"""Add the per-plan feature-assignment table to the template context."""
return super().get_context_data(**kwargs) | {
"features_table": FeatureAssignmentTable(
FeatureAssignment.objects.filter(plan=self.object).order_by(
"feature__name"
),
create_url=reverse(
"ferrobase:featureassignment_create",
query={
FeatureAssignmentCreateView.parent_query: str(self.object.pk),
},
),
),
}
[docs]
class SubscriptionPlanDeleteView(DeleteView):
"""Delete a subscription plan."""
model = SubscriptionPlan
subscription_plan_urls = model_crud_routes(
SubscriptionPlan,
read_view=SubscriptionPlanTableView,
copy_view=SubscriptionPlanCopyView,
create_view=SubscriptionPlanCreateView,
update_view=SubscriptionPlanUpdateView,
delete_view=SubscriptionPlanDeleteView,
)