133 lines
5.3 KiB
Python
133 lines
5.3 KiB
Python
import uuid
|
|
from typing import Optional
|
|
from sqlalchemy.orm import Session, joinedload
|
|
from sqlalchemy import or_, cast, String
|
|
from fastapi import HTTPException, status
|
|
|
|
from app.models.auth.subscription_plan_model import SubscriptionPlan
|
|
from app.models.auth.plan_access_model import PlanAccess
|
|
from app.models.auth.plan_module_access_model import PlanModuleAccess
|
|
from app.schemas.auth.subscription_plan_schema import (
|
|
SubscriptionPlanCreate,
|
|
SubscriptionPlanUpdate,
|
|
SubscriptionPlanPaginatedResponse,
|
|
SubscriptionPlanResponse
|
|
)
|
|
|
|
class SubscriptionPlanService:
|
|
|
|
@staticmethod
|
|
def create_plan(db: Session, plan_data: SubscriptionPlanCreate) -> SubscriptionPlan:
|
|
existing = db.query(SubscriptionPlan).filter(SubscriptionPlan.name == plan_data.name).first()
|
|
if existing:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Plan name already exists")
|
|
|
|
plan = SubscriptionPlan(
|
|
name=plan_data.name,
|
|
description=plan_data.description,
|
|
price=plan_data.price,
|
|
is_public=plan_data.is_public,
|
|
status=plan_data.status
|
|
)
|
|
db.add(plan)
|
|
db.flush()
|
|
|
|
if plan_data.access_ids:
|
|
for acc_id in plan_data.access_ids:
|
|
db.add(PlanAccess(plan_id=plan.id, access_id=acc_id))
|
|
|
|
if plan_data.module_access_ids:
|
|
for macc_id in plan_data.module_access_ids:
|
|
db.add(PlanModuleAccess(plan_id=plan.id, module_access_id=macc_id))
|
|
|
|
db.commit()
|
|
db.refresh(plan)
|
|
return plan
|
|
|
|
@staticmethod
|
|
def update_plan(db: Session, plan_id: uuid.UUID, plan_data: SubscriptionPlanUpdate) -> SubscriptionPlan:
|
|
plan = db.query(SubscriptionPlan).filter(SubscriptionPlan.id == plan_id).first()
|
|
if not plan:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Plan not found")
|
|
|
|
update_dict = plan_data.model_dump(exclude_unset=True)
|
|
|
|
if "name" in update_dict and update_dict["name"] != plan.name:
|
|
existing = db.query(SubscriptionPlan).filter(SubscriptionPlan.name == update_dict["name"]).first()
|
|
if existing:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Plan name already exists")
|
|
|
|
if "access_ids" in update_dict:
|
|
access_ids = update_dict.pop("access_ids")
|
|
db.query(PlanAccess).filter(PlanAccess.plan_id == plan.id).delete()
|
|
if access_ids:
|
|
for acc_id in access_ids:
|
|
db.add(PlanAccess(plan_id=plan.id, access_id=acc_id))
|
|
|
|
if "module_access_ids" in update_dict:
|
|
module_access_ids = update_dict.pop("module_access_ids")
|
|
db.query(PlanModuleAccess).filter(PlanModuleAccess.plan_id == plan.id).delete()
|
|
if module_access_ids:
|
|
for macc_id in module_access_ids:
|
|
db.add(PlanModuleAccess(plan_id=plan.id, module_access_id=macc_id))
|
|
|
|
for key, value in update_dict.items():
|
|
setattr(plan, key, value)
|
|
|
|
db.commit()
|
|
db.refresh(plan)
|
|
return plan
|
|
|
|
@staticmethod
|
|
def get_plan(db: Session, plan_id: uuid.UUID) -> SubscriptionPlan:
|
|
plan = (
|
|
db.query(SubscriptionPlan)
|
|
.options(
|
|
joinedload(SubscriptionPlan.plan_accesses),
|
|
joinedload(SubscriptionPlan.plan_module_accesses)
|
|
)
|
|
.filter(SubscriptionPlan.id == plan_id)
|
|
.first()
|
|
)
|
|
if not plan:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Plan not found")
|
|
return plan
|
|
|
|
@staticmethod
|
|
def get_all_plans(db: Session, is_public: Optional[bool] = None, status: Optional[str] = None):
|
|
query = db.query(SubscriptionPlan)
|
|
if is_public is not None:
|
|
query = query.filter(SubscriptionPlan.is_public == is_public)
|
|
if status is not None:
|
|
query = query.filter(SubscriptionPlan.status == status)
|
|
return query.all()
|
|
|
|
@staticmethod
|
|
def get_paginated_plans(db: Session, page: int = 1, page_size: int = 10, search: Optional[str] = None) -> SubscriptionPlanPaginatedResponse:
|
|
query = db.query(SubscriptionPlan)
|
|
if search:
|
|
query = query.filter(
|
|
or_(
|
|
SubscriptionPlan.name.ilike(f"%{search}%"),
|
|
cast(SubscriptionPlan.id, String).ilike(f"%{search}%")
|
|
)
|
|
)
|
|
total = query.count()
|
|
offset = (page - 1) * page_size
|
|
plans = query.offset(offset).limit(page_size).all()
|
|
total_pages = (total + page_size - 1) // page_size if total > 0 else 0
|
|
|
|
return SubscriptionPlanPaginatedResponse(
|
|
items=[SubscriptionPlanResponse.from_orm(p) for p in plans],
|
|
total=total,
|
|
page=page,
|
|
page_size=page_size,
|
|
total_pages=total_pages
|
|
)
|
|
|
|
@staticmethod
|
|
def delete_plan(db: Session, plan_id: uuid.UUID):
|
|
plan = SubscriptionPlanService.get_plan(db, plan_id)
|
|
db.delete(plan)
|
|
db.commit()
|
|
return {"message": "Plan deleted successfully"} |