Replies: 2 comments 1 reply
-
# approval_service.py
from models import db, Approval, User, ApprovalStatus
from utils import current_time
import logging
logger = logging.getLogger(__name__)
class ApprovalService:
@staticmethod
def get_approval(model, instance_id, field_name):
logger.info(f"Getting approval for {model.__name__} {instance_id} {field_name}")
return Approval.query.filter_by(
content_type=model.__name__,
content_id=instance_id,
field_name=field_name,
status=ApprovalStatus.PENDING # 使用枚举值
).order_by(Approval.submit_time.desc()).first()
@staticmethod
def has_pending_approval(model, instance_id, field_name):
return ApprovalService.get_approval(model, instance_id, field_name) is not None
@staticmethod
def get_current_value(model, instance_id, field_name):
instance = model.query.get(instance_id)
approval = ApprovalService.get_approval(model, instance_id, field_name)
if approval:
return None # 或者返回一个表示"待审核"的值
return getattr(instance, field_name)
@staticmethod
def submit_for_approval(model, instance_id, field_name, new_value, submitter_id):
logger.info(f"Submitting for approval {model.__name__} {instance_id} {field_name} {new_value} {submitter_id}")
if ApprovalService.has_pending_approval(model, instance_id, field_name):
logger.warning(f"Already has pending approval for {model.__name__} {instance_id} {field_name}")
return False, f"已有待审核的{field_name}"
approval = Approval(
content_type=model.__name__,
content_id=instance_id,
submitter_id=submitter_id,
field_name=field_name,
new_value=new_value,
status=ApprovalStatus.PENDING
)
db.session.add(approval)
try:
db.session.commit()
logger.info(f"Successfully submitted for approval {model.__name__} {instance_id} {field_name}")
return True, f"{field_name}已提交审核"
except Exception as e:
db.session.rollback()
logger.error(f"Error submitting for approval {model.__name__} {instance_id} {field_name}: {str(e)}")
return False, f"提交审核时发生错误: {str(e)}"
@staticmethod
def approve(approval_id, reviewer_id, comment=None):
logger.info(f"Approving approval {approval_id} {reviewer_id} {comment}")
approval = Approval.query.get(approval_id)
if not approval:
logger.error(f"Approval record not found for ID: {approval_id}")
return False, "审核记录不存在"
approval.status = ApprovalStatus.APPROVED # 使用枚举值
approval.reviewer_id = reviewer_id
approval.review_time = current_time()
approval.review_comment = comment
if approval.content_type == 'User' and approval.field_name == 'avatar':
user = User.query.get(approval.content_id)
user.avatar = approval.new_value
db.session.commit()
return True, "审核已通过"
@staticmethod
def reject(approval_id, reviewer_id, comment=None):
approval = Approval.query.get(approval_id)
if not approval:
return False, "审核记录不存在"
approval.status = ApprovalStatus.REJECTED # 使用枚举值
approval.reviewer_id = reviewer_id
approval.review_time = current_time()
approval.review_comment = comment
db.session.commit()
return True, "审核已拒绝" |
Beta Was this translation helpful? Give feedback.
0 replies
-
|
the code listing is missing them import for i strongly suspect that switching to enum.StrEnum will fix this and recommend trying that hypothesis |
Beta Was this translation helpful? Give feedback.
1 reply
-
|
Already imported PyEnum in the model, but I'll try enum.StrEnum too! |
Beta Was this translation helpful? Give feedback.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
-
I've asked various AIs about this issue and nothing has been resolved, here's a request for help
my model：
After running Flask, an error message is displayed.
Beta Was this translation helpful? Give feedback.
All reactions