fixing flake8 comments

met-api/src/met_api/resources/widget_poll.py

@@ -38,7 +38,7 @@ def post(widget_id):
         """Create poll widget."""
         try:
             request_json = request.get_json()
-            valid_format, errors = Polls.validate_response_format(request_json)
+            valid_format, errors = Polls.validate_request_format(request_json)
             if not valid_format:
                 return {'message': 'Invalid response format', 'errors': errors}, HTTPStatus.BAD_REQUEST
             widget_poll = WidgetPollService().create_poll(widget_id, request_json)
@@ -47,7 +47,8 @@ def post(widget_id):
             return str(err), err.status_code
 
     @staticmethod
-    def validate_response_format(data):
+    def validate_request_format(data):
+        """Validate response format."""
         valid_format, errors = schema_utils.validate(data, 'poll_widget')
         if not valid_format:
             errors = schema_utils.serialize(errors)
@@ -64,10 +65,9 @@ class Poll(Resource):
     @_jwt.requires_auth
     def patch(widget_id, poll_widget_id):
         """Update poll widget."""
-
         try:
             request_json = request.get_json()
-            valid_format, errors = Poll.validate_response_format(request_json)
+            valid_format, errors = Poll.validate_request_format(request_json)
             if not valid_format:
                 return {'message': 'Invalid response format', 'errors': errors}, HTTPStatus.BAD_REQUEST
 
@@ -77,12 +77,14 @@ def patch(widget_id, poll_widget_id):
             return str(err), err.status_code
 
     @staticmethod
-    def validate_response_format(data):
+    def validate_request_format(data):
+        """Validate request format."""
         valid_format, errors = schema_utils.validate(data, 'poll_widget_update')
         if not valid_format:
             errors = schema_utils.serialize(errors)
         return valid_format, errors
 
+
 @cors_preflight('POST')
 @API.route('/<int:poll_widget_id>/responses')
 class PollResponseRecord(Resource):
@@ -94,7 +96,7 @@ def post(widget_id, poll_widget_id):
         """Record a response for a given poll widget."""
         try:
             response_data = request.get_json()
-            valid_format, errors = PollResponseRecord.validate_response_format(response_data)
+            valid_format, errors = PollResponseRecord.validate_request_format(response_data)
             if not valid_format:
                 return {'message': 'Invalid response format', 'errors': errors}, HTTPStatus.BAD_REQUEST
 
@@ -112,14 +114,16 @@ def post(widget_id, poll_widget_id):
             return err.error, err.status_code
 
     @staticmethod
-    def validate_response_format(data):
+    def validate_request_format(data):
+        """Validate Request format."""
         valid_format, errors = schema_utils.validate(data, 'poll_response')
         if not valid_format:
             errors = schema_utils.serialize(errors)
         return valid_format, errors
 
     @staticmethod
     def prepare_response_data(data, widget_id, poll_widget_id):
+        """Prepare poll response object."""
         response_dict = dict(data)
         response_dict['poll_id'] = poll_widget_id
         response_dict['widget_id'] = widget_id
@@ -128,14 +132,17 @@ def prepare_response_data(data, widget_id, poll_widget_id):
 
     @staticmethod
     def is_poll_active(poll_id):
+        """Check if poll active or not."""
         return WidgetPollService.is_poll_active(poll_id)
 
     @staticmethod
     def is_poll_limit_exceeded(poll_id, participant_id):
+        """Check poll limt execeeded or not."""
         return WidgetPollService.check_already_polled(poll_id, participant_id, 10)
 
     @staticmethod
     def record_poll_response(response_dict):
+        """Record poll respinse in database."""
         poll_response = WidgetPollService.record_response(response_dict)
         if poll_response.id:
             return {'message': 'Response recorded successfully'}, HTTPStatus.CREATED

met-api/src/met_api/schemas/widget_poll.py

@@ -1,17 +1,42 @@
-"""Schema for WidgetPoll."""
+"""Schema for Widget Poll."""
 from met_api.models.widget_poll import Poll as PollModel
 from met_api.models.poll_answers import PollAnswer as PollAnswerModel
 from marshmallow import Schema
 from marshmallow_sqlalchemy.fields import Nested
 
+
 class PollAnswerSchema(Schema):
+    """
+    Schema for serializing and deserializing Poll Answer data.
+
+    This schema is used to represent poll answers in a structured format,
+    facilitating operations like loading from and dumping to JSON.
+    """
+
     class Meta:
-        model = PollAnswerModel
-        fields = ('id', 'answer_text', 'poll_id')
+        """Meta class for PollAnswerSchema options."""
+
+        model = PollAnswerModel  # The model representing Poll Answer.
+        fields = ('id', 'answer_text', 'poll_id')  # Fields to include in the schema.
+
 
 class WidgetPollSchema(Schema):
+    """
+    Schema for serializing and deserializing Widget Poll data.
+
+    This schema is designed to handle Widget Poll data, enabling easy conversion
+    between Python objects and JSON representation, specifically for Widget Polls.
+    """
+
     class Meta:
-        model = PollModel
+        """Meta class for WidgetPollSchema options."""
+
+        model = PollModel  # The model representing Widget Poll.
         fields = ('id', 'title', 'description', 'status', 'widget_id', 'engagement_id', 'answers')
 
     answers = Nested(PollAnswerSchema, many=True)
+    """Nested field for Poll Answers.
+
+    This field represents a collection of Poll Answers associated with a Widget Poll,
+    allowing for the inclusion of related Poll Answer data within a Widget Poll's serialized form.
+    """

met-api/src/met_api/services/poll_answers_service.py

@@ -1,6 +1,7 @@
 """Service for PollAnswer management."""
 from http import HTTPStatus
 
+from sqlalchemy.exc import SQLAlchemyError
 from met_api.exceptions.business_exception import BusinessException
 from met_api.models.poll_answers import PollAnswer as PollAnswerModel
 
@@ -20,7 +21,7 @@ def create_bulk_poll_answers(poll_id: int, answers_data: list):
         try:
             if len(answers_data) > 0:
                 PollAnswerModel.bulk_insert_answers(poll_id, answers_data)
-        except Exception as e:
+        except SQLAlchemyError as e:
             raise BusinessException(str(e), HTTPStatus.INTERNAL_SERVER_ERROR) from e
 
     @staticmethod

met-api/src/met_api/services/poll_response_service.py

@@ -1,4 +1,7 @@
 """Service for Poll Response management."""
+from http import HTTPStatus
+from sqlalchemy.exc import SQLAlchemyError
+from met_api.exceptions.business_exception import BusinessException
 from met_api.models.poll_responses import PollResponse as PollResponseModel
 from met_api.services.poll_answers_service import PollAnswerService
 
@@ -10,6 +13,7 @@ class PollResponseService:
     def create_response(response_data: dict) -> PollResponseModel:
         """
         Create a poll response.
+
         Raises ValueError if the selected answer is not valid for the poll.
         """
         try:
@@ -19,25 +23,28 @@ def create_response(response_data: dict) -> PollResponseModel:
             # Validate if the poll and answer are valid
             valid_answers = PollAnswerService.get_poll_answer(poll_id)
             if not any(answer.id == selected_answer_id for answer in valid_answers):
-                raise ValueError('Invalid selected answer for the poll.')
+                raise BusinessException('Invalid selected answer for the poll.', HTTPStatus.BAD_REQUEST)
 
             # Create and save the poll response
             poll_response = PollResponseModel(**response_data)
             poll_response.save()
             return poll_response
-        except Exception as e:
+        except SQLAlchemyError as e:
             # Log the exception or handle it as needed
-            raise ValueError(f'Error creating poll response: {e}') from e
+            raise BusinessException(f'Error creating poll response: {e}',
+                                    HTTPStatus.INTERNAL_SERVER_ERROR) from e
 
     @staticmethod
     def get_poll_count(poll_id: int, ip_addr: str = None) -> int:
         """
         Get the count of responses for a given poll.
+
         Optionally filters by participant IP.
         """
         try:
             responses = PollResponseModel.get_responses_by_participant_id(poll_id, ip_addr)
             return len(responses)
-        except Exception as e:
+        except SQLAlchemyError as e:
             # Log the exception or handle it as needed
-            raise ValueError(f'Error creating poll response: {e}') from e
+            raise BusinessException(f'Error creating poll response: {e}',
+                                    HTTPStatus.INTERNAL_SERVER_ERROR) from e

met-api/src/met_api/services/widget_poll_service.py

@@ -1,6 +1,7 @@
 """Service for Widget Poll management."""
 from http import HTTPStatus
 
+from sqlalchemy.exc import SQLAlchemyError
 from met_api.constants.membership_type import MembershipType
 from met_api.exceptions.business_exception import BusinessException
 from met_api.models.widget_poll import Poll as PollModel
@@ -33,7 +34,7 @@ def create_poll(widget_id: int, poll_details: dict):
             eng_id = poll_details.get('engagement_id')
             WidgetPollService._check_authorization(eng_id)
             return WidgetPollService._create_poll_model(widget_id, poll_details)
-        except Exception as exc:
+        except SQLAlchemyError as exc:
             raise BusinessException(str(exc), HTTPStatus.BAD_REQUEST) from exc
 
     @staticmethod
@@ -47,15 +48,15 @@ def update_poll(widget_id: int, poll_widget_id: int, poll_data: dict):
                 raise BusinessException('Invalid widget ID', HTTPStatus.BAD_REQUEST)
 
             return WidgetPollService._update_poll_model(poll_widget_id, poll_data)
-        except Exception as exc:
+        except SQLAlchemyError as exc:
             raise BusinessException(str(exc), HTTPStatus.BAD_REQUEST) from exc
 
     @staticmethod
     def record_response(response_data: dict):
         """Record a response for a poll."""
         try:
             return PollResponseService.create_response(response_data)
-        except Exception as exc:
+        except SQLAlchemyError as exc:
             raise BusinessException(str(exc), HTTPStatus.BAD_REQUEST) from exc
 
     @staticmethod
@@ -64,7 +65,7 @@ def check_already_polled(poll_id: int, ip_addr: str, count: int) -> bool:
         try:
             poll_count = PollResponseService.get_poll_count(poll_id, ip_addr)
             return poll_count >= count
-        except Exception as exc:
+        except SQLAlchemyError as exc:
             raise BusinessException(str(exc), HTTPStatus.BAD_REQUEST) from exc
 
     @staticmethod
@@ -73,7 +74,7 @@ def is_poll_active(poll_id: int) -> bool:
         try:
             poll = WidgetPollService.get_poll_by_id(poll_id)
             return poll.status == 'active'
-        except Exception as exc:
+        except SQLAlchemyError as exc:
             raise BusinessException(str(exc), HTTPStatus.BAD_REQUEST) from exc
 
     @staticmethod

met-api/src/met_api/utils/ip_util.py

@@ -1,3 +1,14 @@
+"""
+This module provides utility functions for handling IP addresses in a Flask application.
+
+It includes a function for hashing IP addresses using SHA256, ensuring secure and consistent
+hashing by combining each IP address with the Flask application's secret key. This approach
+is typically used for anonymizing IP addresses while maintaining the ability to identify
+sessions or users without storing their actual IP addresses.
+
+Functions:
+    hash_ip(ip_address): Hashes an IP address with the Flask secret key.
+"""
 from hashlib import sha256
 from flask import current_app
 
@@ -16,4 +27,4 @@ def hash_ip(ip_address):
     secret_key = current_app.config.get('SECRET_KEY', '')
 
     # Concatenate the IP address and secret key, and hash the resulting string
-    return sha256(f"{ip_address}{secret_key}".encode()).hexdigest()
+    return sha256(f'{ip_address}{secret_key}'.encode()).hexdigest()

met-api/tests/unit/api/test_widget_poll.py

@@ -21,17 +21,18 @@
 from http import HTTPStatus
 
 from faker import Faker
-from tests.utilities.factory_scenarios import TestJwtClaims, TestWidgetPollInfo, TestPollAnswerInfo
-from tests.utilities.factory_utils import factory_auth_header, factory_engagement_model, factory_widget_model, \
-    factory_poll_model, factory_poll_answer_model
 
 from met_api.utils.enums import ContentType
+from tests.utilities.factory_scenarios import TestJwtClaims, TestPollAnswerInfo, TestWidgetPollInfo
+from tests.utilities.factory_utils import (
+    factory_auth_header, factory_engagement_model, factory_poll_answer_model, factory_poll_model, factory_widget_model)
+
 
 fake = Faker()
 
 
 def test_get_widget(client, jwt, session):
-    """Assert that a get API endpoint is working as expected"""
+    """Assert that a get API endpoint is working as expected."""
     # Test setup: create a poll widget and a response model
     headers = factory_auth_header(jwt=jwt, claims=TestJwtClaims.no_role)
     engagement = factory_engagement_model()
@@ -54,7 +55,6 @@ def test_get_widget(client, jwt, session):
     assert json_data[0]['answers'][0]['answer_text'] == answer.answer_text
 
 
-
 def test_create_poll_widget(client, jwt, session, setup_admin_user_and_claims):
     """Assert that a poll widget can be POSTed."""
     # Test setup: create a poll widget model
@@ -99,7 +99,7 @@ def test_create_poll_widget(client, jwt, session, setup_admin_user_and_claims):
 
     # Sending POST request
     rv = client.post(
-        f'/api/widgets/100/polls',
+        '/api/widgets/100/polls',
         data=json.dumps(data),
         headers=headers,
         content_type=ContentType.JSON.value,
@@ -151,7 +151,7 @@ def test_update_poll_widget(client, jwt, session, setup_admin_user_and_claims):
     assert rv.status_code == HTTPStatus.BAD_REQUEST
 
 
-def test_record_poll_response(client, jwt, session):
+def test_record_poll_response(client, session, jwt):
     """Assert that a response for a poll widget can be POSTed."""
     # Test setup: create a poll widget and a response model
     headers = factory_auth_header(jwt=jwt, claims=TestJwtClaims.no_role)

met-api/tests/unit/models/test_engagement.py

@@ -20,10 +20,11 @@
 
 from met_api.constants.engagement_status import Status
 from met_api.models.engagement import Engagement as EngagementModel
-from met_api.models.pagination_options import PaginationOptions
 from met_api.models.engagement_scope_options import EngagementScopeOptions
+from met_api.models.pagination_options import PaginationOptions
 from tests.utilities.factory_utils import factory_engagement_model
 
+
 fake = Faker()
 
 

met-api/tests/unit/models/test_poll_answer.py

@@ -16,18 +16,17 @@
 Test suite to ensure that the Engagement model routines are working as expected.
 """
 
-from tests.utilities.factory_scenarios import TestPollAnswerInfo
-from tests.utilities.factory_utils import factory_poll_model, factory_poll_answer_model, factory_engagement_model, \
-    factory_widget_model
-
 from met_api.models.poll_answers import PollAnswer
+from tests.utilities.factory_scenarios import TestPollAnswerInfo
+from tests.utilities.factory_utils import (
+    factory_engagement_model, factory_poll_answer_model, factory_poll_model, factory_widget_model)
 
 
 def test_get_answers(session):
     """Assert that answers for a poll can be fetched."""
     poll = _create_poll()
-    answer1 = factory_poll_answer_model(poll, TestPollAnswerInfo.answer1)
-    answer2 = factory_poll_answer_model(poll, TestPollAnswerInfo.answer2)
+    factory_poll_answer_model(poll, TestPollAnswerInfo.answer1)
+    factory_poll_answer_model(poll, TestPollAnswerInfo.answer2)
     session.commit()
     answers = PollAnswer.get_answers(poll.id)
     assert len(answers) == 2
@@ -64,13 +63,13 @@ def test_bulk_insert_answers(session):
 
 
 def _create_poll():
-    """Helper function to create a poll for testing."""
+    """Create and return a sample poll for testing."""
     widget = _create_widget()
     return factory_poll_model(widget, {'title': 'Sample Poll', 'engagement_id': widget.engagement_id})
 
 
 def _create_widget():
-    """Helper function to create a widget for testing."""
+    """Create and return a sample widget for testing."""
     engagement = factory_engagement_model()
     widget = factory_widget_model({'engagement_id': engagement.id})
     return widget