7bbe4e7295
- Added subscription_service.py for managing subscriptions with functions to create, read, update, and delete subscriptions. - Added tenant_service.py for managing tenants with functions to create, read, update, delete tenants, and assign users to tenants. - Added user_service.py for managing users with functions to create, read, update, and delete users. - Created tests for business types, plans, subscriptions, tenants, and users, ensuring proper functionality and error handling. - Implemented pagination in user listing and included user profiles in responses. - Established fixtures for database setup and teardown in tests.
75 lines
2.1 KiB
Python
75 lines
2.1 KiB
Python
from datetime import datetime
|
|
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.plan import Plan
|
|
from app.schemas.plan import PlanCreate, PlanUpdate
|
|
|
|
|
|
def get_plan_list(db: Session):
|
|
return db.query(Plan).filter(Plan.deleted_at.is_(None)).all()
|
|
|
|
|
|
def get_plan_by_id(db: Session, id: str):
|
|
plan = db.query(Plan).filter(Plan.id == id, Plan.deleted_at.is_(None)).first()
|
|
if not plan:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail={"message": "The item does not exist"},
|
|
)
|
|
return plan
|
|
|
|
|
|
def create_plan(db: Session, data: PlanCreate):
|
|
existing = db.query(Plan).filter(Plan.code == data.code).first()
|
|
if existing:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail={"message": "Code already exists"},
|
|
)
|
|
|
|
plan = Plan(**data.model_dump())
|
|
db.add(plan)
|
|
db.commit()
|
|
db.refresh(plan)
|
|
return plan
|
|
|
|
|
|
def update_plan(db: Session, id: str, data: PlanUpdate):
|
|
plan = db.query(Plan).filter(Plan.id == id, Plan.deleted_at.is_(None)).first()
|
|
if not plan:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail={"message": "The item does not exist"},
|
|
)
|
|
|
|
update_data = data.model_dump(exclude_unset=True)
|
|
|
|
if "code" in update_data:
|
|
dup = db.query(Plan).filter(Plan.code == update_data["code"], Plan.id != id).first()
|
|
if dup:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail={"message": "Code already exists"},
|
|
)
|
|
|
|
for field, value in update_data.items():
|
|
setattr(plan, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(plan)
|
|
return plan
|
|
|
|
|
|
def delete_plan(db: Session, id: str):
|
|
plan = db.query(Plan).filter(Plan.id == id, Plan.deleted_at.is_(None)).first()
|
|
if not plan:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail={"message": "The item does not exist"},
|
|
)
|
|
|
|
plan.deleted_at = datetime.now()
|
|
db.commit()
|