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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
-
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