diff --git a/src/api/handlers/campaign.py b/src/api/handlers/campaign.py index 909d30ce4..b0ced4b4a 100644 --- a/src/api/handlers/campaign.py +++ b/src/api/handlers/campaign.py @@ -29,6 +29,7 @@ def registerRoutes(self): self.add("/campaigns.technologies.testimonials.create", self.create_campaign_technology_testimonial) self.add("/campaigns.technologies.testimonials.update", self.update_campaign_technology_testimonial) + self.add("/campaigns.technologies.testimonials.delete", self.delete_campaign_technology_testimonial) self.add("/campaigns.technologies.comments.create", self.create_campaign_technology_comment) self.add("/campaigns.technologies.comments.update", self.update_campaign_technology_comment) @@ -620,6 +621,22 @@ def add_campaign_technology_view(self, request): if err: return err return MassenergizeResponse(data=res) + + + + def delete_campaign_technology_testimonial(self, request): + context: Context = request.context + args: dict = context.args + + self.validator.expect("id", str, is_required=True) + args, err = self.validator.verify(args) + if err: + return err + + res, err = self.service.delete_campaign_technology_testimonial(context, args) + if err: + return err + return MassenergizeResponse(data=res) diff --git a/src/api/handlers/technology.py b/src/api/handlers/technology.py new file mode 100644 index 000000000..cea137271 --- /dev/null +++ b/src/api/handlers/technology.py @@ -0,0 +1,292 @@ + + + +from _main_.utils.context import Context +from _main_.utils.massenergize_response import MassenergizeResponse +from _main_.utils.route_handler import RouteHandler +from api.services.technology import TechnologyService + + +class TechnologyHandler(RouteHandler): + def __init__(self): + super().__init__() + self.service = TechnologyService() + self.registerRoutes() + + def registerRoutes(self): + self.add("/technologies.info", self.info) + self.add("/technologies.create", self.create) + self.add("/technologies.list", self.list) + self.add("/technologies.update", self.update) + self.add("/technologies.delete", self.delete) + self.add("/technologies.coaches.add", self.add_coach) + self.add("/technologies.coaches.remove", self.remove_coach) + self.add("/technologies.listForAdmin", self.list_for_admin) + self.add("/technologies.vendors.add", self.add_vendor) + self.add("/technologies.vendors.remove", self.remove_vendor) + self.add("/technologies.vendors.list", self.list_vendors) + self.add("/technologies.overview.create", self.create_overview) + self.add("/technologies.overview.update", self.update_overview) + self.add("/technologies.overview.delete", self.delete_overview) + self.add("/technologies.overview.list", self.list_overviews) + + + + def info(self, request): + context: Context = request.context + args: dict = context.args + + self.validator.expect("id", str, is_required=True) + + args, err = self.validator.verify(args, strict=True) + if err: + return err + + res, err = self.service.get_technology_info(context, args) + if err: + return err + return MassenergizeResponse(data=res) + + + def list(self, request): + context: Context = request.context + args: dict = context.args + + res, err = self.service.list_technologies(context, args) + if err: + return err + return MassenergizeResponse(data=res) + + + + def create(self, request): + context: Context = request.context + args: dict = context.args + + self.validator.expect("name", str, is_required=True) + self.validator.expect("description", str, is_required=True) + self.validator.expect("image", str, is_required=True) + + args, err = self.validator.verify(args, strict=True) + + if err: + return err + + res, err = self.service.create_technology(context, args) + if err: + return err + return MassenergizeResponse(data=res) + + + def update(self, request): + context: Context = request.context + args: dict = context.args + + self.validator.expect("id", str, is_required=True) + self.validator.expect("name", str, is_required=False) + self.validator.expect("description", str, is_required=False) + self.validator.expect("image", str, is_required=False) + + + args, err = self.validator.verify(args, strict=True) + + if err: + return err + + res, err = self.service.update_technology(context, args) + if err: + return err + return MassenergizeResponse(data=res) + + + def delete(self, request): + context: Context = request.context + args: dict = context.args + + self.validator.expect("id", str, is_required=True) + + args, err = self.validator.verify(args, strict=True) + + if err: + return err + + res, err = self.service.delete_technology(context, args) + if err: + return err + return MassenergizeResponse(data=res) + + + + def list_for_admin(self, request): + context: Context = request.context + args: dict = context.args + + res, err = self.service.list_technologies_for_admin(context, args) + if err: + return err + return MassenergizeResponse(data=res) + + + def add_coach(self, request): + context: Context = request.context + args: dict = context.args + + self.validator.expect("technology_id", str, is_required=True) + self.validator.expect("user_id", str, is_required=True) + + args, err = self.validator.verify(args, strict=True) + + if err: + return err + + res, err = self.service.add_technology_coach(context, args) + if err: + return err + return MassenergizeResponse(data=res) + + + def remove_coach(self, request): + context: Context = request.context + args: dict = context.args + + self.validator.expect("technology_id", str, is_required=True) + self.validator.expect("user_id", str, is_required=True) + + args, err = self.validator.verify(args, strict=True) + + if err: + return err + + res, err = self.service.remove_technology_coach(context, args) + if err: + return err + return MassenergizeResponse(data=res) + + + def add_vendor(self, request): + context: Context = request.context + args: dict = context.args + + self.validator.expect("technology_id", str, is_required=True) + self.validator.expect("vendor_id", str, is_required=True) + + args, err = self.validator.verify(args, strict=True) + + if err: + return err + + res, err = self.service.add_technology_vendor(context, args) + if err: + return err + return MassenergizeResponse(data=res) + + + def remove_vendor(self, request): + context: Context = request.context + args: dict = context.args + + self.validator.expect("technology_id", str, is_required=True) + self.validator.expect("vendor_id", str, is_required=True) + + args, err = self.validator.verify(args, strict=True) + + if err: + return err + + res, err = self.service.remove_technology_vendor(context, args) + if err: + return err + return MassenergizeResponse(data=res) + + + def create_overview(self, request): + context: Context = request.context + args: dict = context.args + + self.validator.expect("technology_id", str, is_required=True) + self.validator.expect("title", str, is_required=True) + self.validator.expect("description", str, is_required=True) + self.validator.expect("image", str, is_required=True) + + args, err = self.validator.verify(args, strict=True) + + if err: + return err + + res, err = self.service.create_technology_overview(context, args) + if err: + return err + return MassenergizeResponse(data=res) + + + def update_overview(self, request): + context: Context = request.context + args: dict = context.args + + self.validator.expect("id", str, is_required=True) + self.validator.expect("title", str, is_required=False) + self.validator.expect("description", str, is_required=False) + self.validator.expect("image", str, is_required=False) + + args, err = self.validator.verify(args, strict=True) + + if err: + return err + + res, err = self.service.update_technology_overview(context, args) + if err: + return err + return MassenergizeResponse(data=res) + + + def delete_overview(self, request): + context: Context = request.context + args: dict = context.args + + self.validator.expect("id", str, is_required=True) + + args, err = self.validator.verify(args, strict=True) + + if err: + return err + + res, err = self.service.delete_technology_overview(context, args) + if err: + return err + return MassenergizeResponse(data=res) + + + + def list_overviews(self, request): + context: Context = request.context + args: dict = context.args + + self.validator.expect("technology_id", str, is_required=True) + + args, err = self.validator.verify(args, strict=True) + + if err: + return err + + res, err = self.service.list_technology_overviews(context, args) + if err: + return err + return MassenergizeResponse(data=res) + + + def list_vendors(self, request): + context: Context = request.context + args: dict = context.args + + self.validator.expect("technology_id", str, is_required=True) + + args, err = self.validator.verify(args, strict=True) + + if err: + return err + + res, err = self.service.list_technology_vendors(context, args) + if err: + return err + return MassenergizeResponse(data=res) + diff --git a/src/api/services/campaign.py b/src/api/services/campaign.py index 40ce7a1f4..f711aeed9 100644 --- a/src/api/services/campaign.py +++ b/src/api/services/campaign.py @@ -16,7 +16,7 @@ def __init__(self): self.store = CampaignStore() def get_campaign_info(self, context: Context, args) -> Tuple[dict, MassEnergizeAPIError]: - campaign, err = self.store.get_Campaign_info(context, args) + campaign, err = self.store.get_campaign_info(context, args) if err: return None, err return serialize(campaign, full=True), None @@ -231,3 +231,11 @@ def add_campaign_technology_like(self, context, args)-> Tuple[dict, MassEnergize if err: return None, err return serialize(res, full=True), None + + + def delete_campaign_technology_testimonial(self, context, args)-> Tuple[dict, MassEnergizeAPIError]: + res, err = self.store.delete_campaign_technology_testimonial(context, args) + if err: + return None, err + + return serialize(res, full=True), None diff --git a/src/api/services/technology.py b/src/api/services/technology.py new file mode 100644 index 000000000..2062084b2 --- /dev/null +++ b/src/api/services/technology.py @@ -0,0 +1,118 @@ + + +from _main_.utils.common import serialize, serialize_all +from _main_.utils.context import Context +from _main_.utils.massenergize_errors import MassEnergizeAPIError +from typing import Tuple + +from api.store.technology import TechnologyStore + + +class TechnologyService: + """ + Service Layer for all the technologies + """ + + def __init__(self): + self.store = TechnologyStore() + + def get_technology_info(self, context: Context, args) -> Tuple[dict, MassEnergizeAPIError]: + res, err = self.store.get_technology_info(context, args) + if err: + return None, err + return serialize(res, full=True), None + + def list_technologies(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + res, err = self.store.list_technologies(context, args) + if err: + return None, err + return serialize_all(res, full=True), None + + def create_technology(self, context: Context, args) -> Tuple[dict, MassEnergizeAPIError]: + res, err = self.store.create_technology(context, args) + if err: + return None, err + + return serialize(res, full=True), None + + + def update_technology(self, context: Context, args) -> Tuple[dict, MassEnergizeAPIError]: + res, err = self.store.update_technology(context, args) + if err: + return None, err + return serialize(res, full=True), None + + + def delete_technology(self, context: Context, args) -> Tuple[dict, MassEnergizeAPIError]: + technology, err = self.store.delete_technology(context, args) + if err: + return None, err + return serialize(technology), None + + + def list_technologies_for_admin(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + technologies, err = self.store.list_technologies_for_admin(context, args) + if err: + return None, err + return serialize_all(technologies), None + + def add_technology_coach(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + res, err = self.store.add_technology_coach(context, args) + if err: + return None, err + return serialize(res), None + + def remove_technology_coach(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + res, err = self.store.remove_technology_coach(context, args) + if err: + return None, err + return serialize(res), None + + def add_technology_vendor(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + res, err = self.store.add_technology_vendor(context, args) + if err: + return None, err + return serialize(res), None + + def remove_technology_vendor(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + res, err = self.store.remove_technology_vendor(context, args) + if err: + return None, err + return serialize(res), None + + def create_technology_overview(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + res, err = self.store.add_technology_overview(context, args) + if err: + return None, err + return serialize(res), None + + def update_technology_overview(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + res, err = self.store.update_technology_overview(context, args) + if err: + return None, err + return serialize(res), None + + def delete_technology_overview(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + res, err = self.store.delete_technology_overview(context, args) + if err: + return None, err + return serialize(res), None + + def list_technology_overviews(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + res, err = self.store.list_technology_overviews(context, args) + if err: + return None, err + return serialize_all(res), None + + def list_technology_vendors(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + res, err = self.store.list_technology_vendors(context, args) + if err: + return None, err + return serialize_all(res), None + + + def list_technologies_for_admin(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + technologies, err = self.store.list_technologies_for_admin(context, args) + if err: + return None, err + return serialize_all(technologies), None \ No newline at end of file diff --git a/src/api/store/campaign.py b/src/api/store/campaign.py index a569ec1df..42b7b65c1 100644 --- a/src/api/store/campaign.py +++ b/src/api/store/campaign.py @@ -14,7 +14,7 @@ class CampaignStore: def __init__(self): self.name = "Campaign Store/DB" - def get_Campaign_info(self, context: Context, args) -> Tuple[dict, MassEnergizeAPIError]: + def get_campaign_info(self, context: Context, args) -> Tuple[dict, MassEnergizeAPIError]: try: campaign_id = args.get("id", None) campaign: Campaign = Campaign.objects.filter(id=campaign_id).first() diff --git a/src/api/store/technology.py b/src/api/store/technology.py new file mode 100644 index 000000000..1786e23ed --- /dev/null +++ b/src/api/store/technology.py @@ -0,0 +1,285 @@ + + +from typing import Tuple + +from sentry_sdk import capture_message +from _main_.utils.context import Context +from _main_.utils.massenergize_errors import CustomMassenergizeError, InvalidResourceError, MassEnergizeAPIError +from apps__campaigns.models import Technology, TechnologyCoach, TechnologyOverview, TechnologyVendor, Vendor +from database.models import Media, UserProfile + + +class TechnologyStore: + def __init__(self): + self.name = "Campaign Store/DB" + + def get_technology_info(self, context: Context, args) -> Tuple[dict, MassEnergizeAPIError]: + try: + technology_id = args.get("id", None) + technology = Technology.objects.filter(id=technology_id).first() + if not technology: + return None, InvalidResourceError() + + return technology, None + + except Exception as e: + capture_message(str(e), level="error") + return None, CustomMassenergizeError(e) + + + def list_technologies(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + try: + technologies = Technology.objects.filter(is_deleted=False) + return technologies, None + except Exception as e: + capture_message(str(e), level="error") + return None, CustomMassenergizeError(e) + + + def create_technology(self, context: Context, args) -> Tuple[dict, MassEnergizeAPIError]: + try: + image = args.pop('image', None) + technology = Technology.objects.create(**args) + if image: + media = Media.objects.create(file=image, name=f"FileUpload for {technology.id} Technology") + technology.image = media + technology.save() + return technology, None + except Exception as e: + capture_message(str(e), level="error") + return None, CustomMassenergizeError(e) + + + def update_technology(self, context: Context, args) -> Tuple[dict, MassEnergizeAPIError]: + try: + technology_id = args.pop('id', None) + image = args.pop('image', None) + technology = Technology.objects.filter(id=technology_id) + if not technology: + return None, InvalidResourceError() + technology.update(**args) + technology = technology.first() + if image: + media = Media.objects.create(file=image, name=f"FileUpload for {technology.id} Technology") + technology.image = media + technology.save() + return technology, None + except Exception as e: + capture_message(str(e), level="error") + return None, CustomMassenergizeError(e) + + + def delete_technology(self, context: Context, args) -> Tuple[dict, MassEnergizeAPIError]: + try: + technology_id = args.get('id', None) + technology = Technology.objects.filter(id=technology_id) + if not technology: + return None, InvalidResourceError() + + technology.update(is_deleted=True) + return technology, None + except Exception as e: + capture_message(str(e), level="error") + return None, CustomMassenergizeError(e) + + + def list_technologies_for_admin(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + try: + technologies = Technology.objects.filter(is_deleted=False) + return technologies, None + except Exception as e: + capture_message(str(e), level="error") + return None, CustomMassenergizeError(e) + + + def add_technology_coach(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + try: + technology_id = args.pop('technology_id', None) + user_id = args.pop('user_id', None) + community = args.pop('community', None) + + technology = Technology.objects.filter(id=technology_id) + if not technology: + return None, InvalidResourceError() + + + coach = TechnologyCoach() + coach.technology = technology.first() + coach.community = community + if user_id: + user = UserProfile.objects.filter(id=user_id).first() + if user: + coach.user = user.first() + + coach.save() + + return coach, None + except Exception as e: + capture_message(str(e), level="error") + return None, CustomMassenergizeError(e) + + + + def remove_technology_coach(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + try: + coach_id = args.pop('technology_coach_id', None) + + if not coach_id: + return None, InvalidResourceError() + + coach = TechnologyCoach.objects.filter(id=coach_id) + if not coach: + return None, CustomMassenergizeError("Invalid Technology Coach ID") + + coach.update(is_deleted=True) + + return coach.first(), None + except Exception as e: + capture_message(str(e), level="error") + return None, CustomMassenergizeError(e) + + + + def add_technology_vendor(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + try: + technology_id = args.pop('technology_id', None) + vendor_id = args.pop('vendor_id', None) + + technology = Technology.objects.filter(id=technology_id) + if not technology: + return None, InvalidResourceError() + + vendor = Vendor.objects.filter(id=vendor_id) + if not vendor: + return None, InvalidResourceError() + + tech_vendor = TechnologyVendor() + tech_vendor.technology = technology.first() + tech_vendor.vendor = vendor.first() + tech_vendor.save() + + return tech_vendor, None + except Exception as e: + capture_message(str(e), level="error") + return None, CustomMassenergizeError(e) + + + + def remove_technology_vendor(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + try: + tech_vendor_id = args.pop('technology_vendor_id', None) + + if not tech_vendor_id: + return None, InvalidResourceError() + + tech_vendor = TechnologyVendor.objects.filter(id=tech_vendor_id) + if not tech_vendor: + return None, CustomMassenergizeError("Invalid Technology Vendor ID") + + tech_vendor.update(is_deleted=True) + + return tech_vendor.first(), None + except Exception as e: + capture_message(str(e), level="error") + return None, CustomMassenergizeError(e) + + + + def list_technology_vendors(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + try: + tech = args.pop("technology_id") + vendors = TechnologyVendor.objects.filter(technology__id=tech,is_deleted=False) + return vendors, None + except Exception as e: + capture_message(str(e), level="error") + return None, CustomMassenergizeError(e) + + + def list_technology_coaches(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + try: + tech = args.pop("technology_id") + coaches = TechnologyCoach.objects.filter(technology__id=tech,is_deleted=False) + return coaches, None + except Exception as e: + capture_message(str(e), level="error") + return None, CustomMassenergizeError(e) + + + def list_technology_overviews(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + try: + tech = args.pop("technology_id") + overviews = TechnologyOverview.objects.filter(technology__id=tech,is_deleted=False) + return overviews, None + except Exception as e: + capture_message(str(e), level="error") + return None, CustomMassenergizeError(e) + + + def add_technology_overview(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + try: + technology_id = args.pop('technology_id', None) + image = args.pop('image', None) + + technology = Technology.objects.filter(id=technology_id) + if not technology: + return None, InvalidResourceError() + args["technology"] = technology.first() + + + tech_overview = TechnologyOverview(**args) + + if image: + media = Media.objects.create(file=image, name=f"FileUpload for {tech_overview.id} TechnologyOverview") + tech_overview.image = media + tech_overview.save() + + return tech_overview, None + except Exception as e: + capture_message(str(e), level="error") + return None, CustomMassenergizeError(e) + + + def update_technology_overview(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + try: + tech_overview_id = args.pop('technology_overview_id', None) + image = args.pop('image', None) + + tech_overview = TechnologyOverview.objects.filter(id=tech_overview_id) + if not tech_overview: + return None, InvalidResourceError() + + if image: + media = Media.objects.create(file=image, name=f"FileUpload for {tech_overview.first().id} TechnologyOverview") + tech_overview.first().image = media + tech_overview.update(**args) + + return tech_overview.first(), None + except Exception as e: + capture_message(str(e), level="error") + return None, CustomMassenergizeError(e) + + + def delete_technology_overview(self, context: Context, args) -> Tuple[list, MassEnergizeAPIError]: + try: + tech_overview_id = args.pop('technology_overview_id', None) + + if not tech_overview_id: + return None, InvalidResourceError() + + tech_overview = TechnologyOverview.objects.filter(id=tech_overview_id) + if not tech_overview: + return None, CustomMassenergizeError("Invalid Technology Overview ID") + + tech_overview.update(is_deleted=True) + + return tech_overview.first(), None + except Exception as e: + capture_message(str(e), level="error") + return None, CustomMassenergizeError(e) + + + + + + +