54 lines
1.8 KiB
Python
54 lines
1.8 KiB
Python
from sqlalchemy.orm import Session
|
|
|
|
from app.common.errors.app_error import NotFoundAppError
|
|
from app.models.entities import PricingRule
|
|
from app.modules.pricing.repository import PricingRepository
|
|
|
|
|
|
class PricingService:
|
|
def __init__(self, db: Session) -> None:
|
|
self.db = db
|
|
self.repository = PricingRepository(db)
|
|
|
|
def list_rules(self) -> list[dict]:
|
|
return [self.serialize(item) for item in self.repository.list_rules().all()]
|
|
|
|
def create_rule(self, payload) -> dict:
|
|
item = PricingRule(**payload.model_dump())
|
|
self.db.add(item)
|
|
self.db.commit()
|
|
self.db.refresh(item)
|
|
return self.serialize(item)
|
|
|
|
def update_rule(self, rule_id: int, payload) -> dict:
|
|
item = self.repository.get_rule(rule_id)
|
|
if not item:
|
|
raise NotFoundAppError("pricing rule not found", code=60004)
|
|
for key, value in payload.model_dump().items():
|
|
setattr(item, key, value)
|
|
self.db.commit()
|
|
return self.serialize(item)
|
|
|
|
def publish_rule(self, rule_id: int) -> dict:
|
|
item = self.repository.get_rule(rule_id)
|
|
if not item:
|
|
raise NotFoundAppError("pricing rule not found", code=60004)
|
|
item.status = 1
|
|
self.db.commit()
|
|
return self.serialize(item)
|
|
|
|
@staticmethod
|
|
def serialize(item: PricingRule) -> dict:
|
|
return {
|
|
"id": item.id,
|
|
"ruleName": item.rule_name,
|
|
"videoModelId": item.video_model_id,
|
|
"pointsPerSecond": item.points_per_second,
|
|
"minimumPoints": item.minimum_points,
|
|
"effectiveAt": item.effective_at.isoformat(),
|
|
"expiredAt": item.expired_at.isoformat() if item.expired_at else None,
|
|
"versionNo": item.version_no,
|
|
"status": item.status,
|
|
}
|
|
|