Untitled
unknown
plain_text
6 months ago
4.9 kB
5
Indexable
from datetime import datetime from typing import List # Helper function for generating field format value def _get_field_format_value(field_detail): if field_detail.type in ["text", "textarea"]: return field_format_mapping.get(field_detail.format) return None # Helper function for handling validation records def _get_validation_ids(session, field_detail): validation_ids = [] validation_record = ( session.query(Validation) .filter_by(type=field_detail.validation_value) .first() ) if validation_record: validation_ids.append(str(validation_record.id)) if field_detail.is_mandatory == "Y": required_validation = ( session.query(Validation) .filter_by(type="Required") .first() ) if required_validation: validation_ids.append(str(required_validation.id)) return ",".join(validation_ids) if validation_ids else None # Helper function for setting default values for field details def _get_default_field_detail_values(current_user): return dict( rcvd_frm_reltio=None, is_dynamic="Y", is_active="Y", is_editable="Y", is_deleted=0, is_visible="Y", is_downstream="Y", is_published="Y", default_option="Y", created_date=datetime.now(), created_by_id=current_user, updated_date=datetime.now(), updated_by_id=current_user, ) # Main mutation function @mutation def create_field_details( self, info: Info, input: List[CreateFieldDetailsInput] ) -> CreateFieldDetailsMutationPayload: try: current_user = info.context.current_user.login created_field_details_ids = [] default_values = _get_default_field_detail_values(current_user) for field_detail in input: field_format_value = _get_field_format_value(field_detail) validation_id = _get_validation_ids(info.context.session, field_detail) merged_field_detail = { **default_values, **{k: v for k, v in field_detail.__dict__.items() if v is not None}, "format": field_format_value, "validation_id": validation_id, "options": field_detail.options if field_detail.type in ["select", "select-multiple"] else None, "rcvd_frm_reltio": field_detail.reltio_id if field_detail.rcvd_frm_reltio == "Y" else None, } new_field_detail = FieldDetail( label=merged_field_detail["label"], name=merged_field_detail["name"], type=merged_field_detail["type"], order=merged_field_detail["order"], format=merged_field_detail["format"], validation_id=merged_field_detail["validation_id"], options=merged_field_detail["options"], rcvd_frm_reltio=merged_field_detail["rcvd_frm_reltio"], is_dynamic=merged_field_detail["is_dynamic"], is_active=merged_field_detail["is_active"], is_editable=merged_field_detail["is_editable"], is_deleted=merged_field_detail["is_deleted"], is_visible=merged_field_detail["is_visible"], is_downstream=merged_field_detail["is_downstream"], is_published=merged_field_detail["is_published"], default_option=merged_field_detail["default_option"], created_date=merged_field_detail["created_date"], created_by_id=merged_field_detail["created_by_id"], updated_date=merged_field_detail["updated_date"], updated_by_id=merged_field_detail["updated_by_id"], ) info.context.session.add(new_field_detail) info.context.session.flush() created_field_details_ids.append(new_field_detail.id) info.context.session.commit() if not _call_stored_procedure(info.context.session): return CreateFieldDetailsMutationPayload( field_details=[], success=False, message=procedure_error ) created_field_details = ( info.context.session.query(FieldDetail) .filter(FieldDetail.id.in_(created_field_details_ids)) .all() ) return CreateFieldDetailsMutationPayload( field_details=created_field_details, success=True, message="Field details created successfully", ) except Exception as e: info.context.session.rollback() return CreateFieldDetailsMutationPayload( field_details=[], success=False, message=f"Creation failed: {str(e)}" )
Editor is loading...
Leave a Comment