Untitled
unknown
plain_text
a month ago
12 kB
7
Indexable
import logging from datetime import datetime from enum import Enum from typing import List, Optional from sqlalchemy import select, update from strawberry import enum, input, mutation, type from app.actions.product_association_helper.product_association_mutation_helper import ( create_product_category, handle_add_submit_type, handle_update_submit_type, is_existing_association, save_product_category, save_subcategories, ) from app.graphql.enums.product_association_submit_type import ( ProductAssociationSubmitTypes, ) from app.actions.category_or_topic_email_trigger.category_or_topic_email_trigger import ( category_admin_distribution_list, product_association_email, send_email, ) from app.graphql.resolvers.product_category_topic_resolver import product_categories from app.config.mail import SMTP_CLIENT from app.graphql.schema import Info from app.graphql.types import UserRoleType, UserType from app.graphql.types.product_category_type import ProductCategoryType from app.models import ( Product, ProductCountry, Group, ProductCategory, ProductCategoryVersion, ProductSubcategory, User, UserRole, ) logger = logging.getLogger(__name__) @type class SaveAssociationResult: updated_associations: List[ProductCategoryType] success: bool message: str = "" @input class ProductAssociationInput: submit_type: ProductAssociationSubmitTypes version_id: Optional[int] = None prod_cat_grp: str mdm_id: int category_id: int subta_1: str sub_category_1: List[int] pro_cat_id: Optional[int] = None @type class DeleteAssociationResult: success: bool message: str = "" @type class ProductAssociationMutation: @mutation async def save_association( self, info: Info, input: ProductAssociationInput ) -> SaveAssociationResult: try: session = info.context.session updated_associations = [] if not input.sub_category_1: return SaveAssociationResult( message="No subcategories provided", success=False, updated_associations=[], ) product_category = create_product_category(input) if is_existing_association(info, input): return SaveAssociationResult( message="Selected combination already exists", success=False, updated_associations=[], ) if input.submit_type == ProductAssociationSubmitTypes.ADD: if not handle_add_submit_type(info, product_category): return SaveAssociationResult( message="No version record found", success=False, updated_associations=[], ) else: if not handle_update_submit_type(info, input, product_category): return SaveAssociationResult( message="Failed to update the last active product category", success=False, updated_associations=[], ) pro_cat_id = save_product_category( info=info, product_category=product_category ) if not pro_cat_id: return SaveAssociationResult( message="Failed to save the product category", success=False, updated_associations=[], ) if not save_subcategories(info, input, pro_cat_id): return SaveAssociationResult( message="Failed to save the product subcategories", success=False, updated_associations=[], ) updated_associations.append(pro_cat_id) topic_names = [] for subcategory_id in input.sub_category_1: topic_name_query = select(Group).where(Group.id == subcategory_id) topic = session.execute(topic_name_query).scalar_one_or_none() if topic: topic_names.append(topic.name) topics_names = ", ".join(topic_names) product_country_query = select(ProductCountry).where(ProductCountry.id == input.mdm_id) product_country = session.execute(product_country_query).scalar_one() distribution_list = await category_admin_distribution_list(session) product_name_query = select(Product).where(Product.id == input.mdm_id) product_name = session.execute(product_name_query).scalar_one() category_name_query = select(Group).where(Group.id == input.category_id) category_name = session.execute(category_name_query).scalar_one() creator_email_query = select(User).where( User.login == info.context.current_user.login ) creator_email = session.execute(creator_email_query).scalar_one() subject = f'INFORM-iMedical MDM - Product "{product_name.product_name}" with Category "{category_name.name}" is saved successfully.' product_association_resolver = await product_association_email(session, input.version_id) if not product_association_resolver: body = (f"Below association saved successfully:<br><br>" f"Functional Group: {input.prod_cat_grp}<br>" f"Product: {product_name.product_name}<br>" f"Category: {category_name.name}<br>" f"Country: {product_country.countries.name}<br>" f"TA Sub Type: {input.subta_1}<br>" f"Topics: {topics_names}<br>") # Run the mailer to execute the email template await send_email( distribution_list, creator_email=creator_email, subject=subject, body=body ) elif product_association_resolver: for resolver in product_association_resolver: resolver_topic_names = [ subcategory.group.name for subcategory in resolver.product_subcategory ] resolver_topics_names = ", ".join(resolver_topic_names) body = (f"Below association saved successfully:<br><br>" f"Functional Group: {input.prod_cat_grp}<br>" f"Product: {product_name.product_name}<br>" f"Category: {category_name.name}<br>" f"Country: {product_country.countries.name}<br>" f"TA Sub Type: {input.subta_1}<br>" f"Topics: {topics_names}<br><br>" f"Previously saved association is given below:<br><br>" f"Functional Group: {product_association_resolver.prod_cat_grp}<br>" f"Product: {product_association_resolver.product_country.product_name}<br>" f"Category: {product_association_resolver.group.name}<br>" f"Country: {product_association_resolver.product_country.country.name}<br>" f"TA Sub Type: {product_association_resolver.ta_sub_type}<br>" f"Topics: {resolver_topics_names}<br>") # Run the mailer to execute the email template await send_email( distribution_list, creator_email=creator_email, subject=subject, body=body ) return SaveAssociationResult( message="Association saved successfully", success=True, updated_associations=updated_associations, ) except Exception as e: return SaveAssociationResult( message=f"An error occurred: {str(e)}", success=False, updated_associations=[], ) @mutation async def delete_association( self, info: Info, input: list[ProductAssociationInput] ) -> DeleteAssociationResult: try: session = info.context.session for row in input: # Fetch the product category by ID product_category = session.get( ProductCategory, row.pro_cat_id ) if not product_category: return DeleteAssociationResult( message="Product category not found", success=False ) # Update the product category to mark it as inactive and deleted product_category.last_active = 0 product_category.is_deleted = 1 product_category.updated_date = datetime.now() product_category.updated_by_id = info.context.current_user.login # Commit the changes to the database session.commit() topic_names = [] for subcategory_id in row.sub_category_1: topic_name_query = select(Group).where(Group.id == subcategory_id) topic = session.execute(topic_name_query).scalar_one_or_none() if topic: topic_names.append(topic.name) topics_names = ", ".join(topic_names) product_country_query = select(ProductCountry).where(ProductCountry.id == row.mdm_id) product_country = session.execute(product_country_query).scalar_one() distribution_list = await category_admin_distribution_list(session) product_name_query = select(Product).where(Product.id == row.mdm_id) product_name = session.execute(product_name_query).scalar_one() category_name_query = select(Group).where(Group.id == row.category_id) category_name = session.execute(category_name_query).scalar_one() creator_email_query = select(User).where( User.login == product_category.created_by_id ) creator_email = session.execute(creator_email_query).scalar_one() subject = f'INFORM-iMedical MDM - Product "{product_name.product_name}" with Category "{category_name.name}" is deleted successfully.' body = (f"Below association has been deleted successfully:<br><br>" f"Functional Group: {product_category.prod_cat_grp}<br>" f"Product: {product_name.product_name}<br>" f"Category: {category_name.name}<br>" f"Country: {product_country.countries.name}<br>" f"TA Sub Type: {product_category.ta_sub_type}<br>" f"Topics: {topics_names}<br>") # Run the mailer to execute the email template await send_email( distribution_list, creator_email=creator_email, subject=subject, body=body ) return DeleteAssociationResult( message="Product association deleted successfully", success=True ) except Exception as e: return DeleteAssociationResult( message=f"An error occurred: {str(e)}", success=False )
Editor is loading...
Leave a Comment