Deseng-463:Poll Widget: Back-end

CHANGELOG.MD

@@ -1,4 +1,8 @@
 ## January 22, 2024
+- **Task** Poll Widget: Back-end [🎟️DESENG-463](https://apps.itsm.gov.bc.ca/jira/browse/DESENG-463)
+  - Created Database models for Widget Poll, Poll Answers, Poll Response.
+  - Created API to manage  Widget Poll, Poll Answers, Poll Response.
+  - Created Unit tests to test the code.
 - **Task** Add missing unit tests for met api [🎟️DESENG-481](https://apps.itsm.gov.bc.ca/jira/browse/DESENG-481)
   - Added missing unit tests for met api
   - Added unit tests for error handling for met api

met-api/migrations/versions/08f69642b7ae_adding_widget_poll.py

@@ -0,0 +1,83 @@
+"""adding_widget_poll
+
+Revision ID: 08f69642b7ae
+Revises: bd0eb0d25caf
+Create Date: 2024-01-16 14:25:07.611485
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision = '08f69642b7ae'
+down_revision = 'bd0eb0d25caf'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('widget_polls',
+    sa.Column('created_date', sa.DateTime(), nullable=False),
+    sa.Column('updated_date', sa.DateTime(), nullable=True),
+    sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+    sa.Column('title', sa.String(length=255), nullable=False),
+    sa.Column('description', sa.String(length=2048), nullable=True),
+    sa.Column('status', sa.Enum('active', 'inactive', name='poll_status'), nullable=True),
+    sa.Column('widget_id', sa.Integer(), nullable=False),
+    sa.Column('engagement_id', sa.Integer(), nullable=False),
+    sa.Column('created_by', sa.String(length=50), nullable=True),
+    sa.Column('updated_by', sa.String(length=50), nullable=True),
+    sa.ForeignKeyConstraint(['engagement_id'], ['engagement.id'], ondelete='CASCADE'),
+    sa.ForeignKeyConstraint(['widget_id'], ['widget.id'], ondelete='CASCADE'),
+    sa.PrimaryKeyConstraint('id')
+    )
+    op.create_table('poll_answers',
+    sa.Column('created_date', sa.DateTime(), nullable=False),
+    sa.Column('updated_date', sa.DateTime(), nullable=True),
+    sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+    sa.Column('answer_text', sa.String(length=255), nullable=False),
+    sa.Column('poll_id', sa.Integer(), nullable=False),
+    sa.Column('created_by', sa.String(length=50), nullable=True),
+    sa.Column('updated_by', sa.String(length=50), nullable=True),
+    sa.ForeignKeyConstraint(['poll_id'], ['widget_polls.id'], ondelete='CASCADE'),
+    sa.PrimaryKeyConstraint('id')
+    )
+    op.create_table('poll_responses',
+    sa.Column('created_date', sa.DateTime(), nullable=False),
+    sa.Column('updated_date', sa.DateTime(), nullable=True),
+    sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+    sa.Column('participant_id', sa.String(length=255), nullable=False),
+    sa.Column('selected_answer_id', sa.Integer(), nullable=False),
+    sa.Column('poll_id', sa.Integer(), nullable=False),
+    sa.Column('widget_id', sa.Integer(), nullable=False),
+    sa.Column('is_deleted', sa.Boolean(), nullable=True),
+    sa.Column('created_by', sa.String(length=50), nullable=True),
+    sa.Column('updated_by', sa.String(length=50), nullable=True),
+    sa.ForeignKeyConstraint(['poll_id'], ['widget_polls.id'], ondelete='CASCADE'),
+    sa.ForeignKeyConstraint(['selected_answer_id'], ['poll_answers.id'], ondelete='CASCADE'),
+    sa.ForeignKeyConstraint(['widget_id'], ['widget.id'], ondelete='CASCADE'),
+    sa.PrimaryKeyConstraint('id')
+    )
+    widget_type_table = sa.table('widget_type',
+        sa.Column('id', sa.Integer),
+        sa.Column('name', sa.String),
+        sa.Column('description', sa.String))
+
+    op.bulk_insert(widget_type_table, [
+        {'id': 10, 'name': 'Poll', 'description': 'The Poll Widget enables real-time polling and feedback collection from public.'}
+    ])
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table('poll_responses')
+    op.drop_table('poll_answers')
+    op.drop_table('widget_polls')
+
+    conn = op.get_bind()
+
+    conn.execute('DELETE FROM widget_type WHERE id=10')
+    # ### end Alembic commands ###

met-api/src/met_api/models/__init__.py

@@ -51,3 +51,6 @@
 from .cac_form import CACForm
 from .widget_timeline import WidgetTimeline
 from .timeline_event import TimelineEvent
+from .widget_poll import Poll
+from .poll_answers import PollAnswer
+from .poll_responses import PollResponse

met-api/src/met_api/models/poll_answers.py

@@ -0,0 +1,56 @@
+"""
+PollAnswers model class.
+
+Manages the Poll answers
+"""
+from __future__ import annotations
+
+from sqlalchemy.sql.schema import ForeignKey
+
+from .base_model import BaseModel
+from .db import db
+
+
+class PollAnswer(BaseModel):
+    """Definition of the PollAnswer entity."""
+
+    __tablename__ = 'poll_answers'
+    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
+    answer_text = db.Column(db.String(255), nullable=False)
+    poll_id = db.Column(db.Integer, ForeignKey('widget_polls.id',
+                                               ondelete='CASCADE'), nullable=False)
+
+    @classmethod
+    def get_answers(cls, poll_id) -> list[PollAnswer]:
+        """Get answers for a poll."""
+        session = db.session.query(PollAnswer)
+        return session.filter(PollAnswer.poll_id == poll_id).all()
+
+    @classmethod
+    def update_answer(cls, answer_id, answer_data: dict) -> PollAnswer:
+        """Update an answer."""
+        answer = PollAnswer.query.get(answer_id)
+        if answer:
+            for key, value in answer_data.items():
+                setattr(answer, key, value)
+            answer.save()
+        return answer
+
+    @classmethod
+    def delete_answers_by_poll_id(cls, poll_id):
+        """Delete answers."""
+        poll_answers = db.session.query(PollAnswer).filter(
+            PollAnswer.poll_id == poll_id
+        )
+        poll_answers.delete()
+        db.session.commit()
+
+    @classmethod
+    def bulk_insert_answers(cls, poll_id, answers):
+        """Bulk insert answers for a poll."""
+        answer_data = [
+            {'poll_id': poll_id, 'answer_text': answer['answer_text']}
+            for answer in answers
+        ]
+        db.session.bulk_insert_mappings(PollAnswer, answer_data)
+        db.session.commit()

met-api/src/met_api/models/poll_responses.py

@@ -0,0 +1,46 @@
+"""
+PollResponse model class.
+
+Manages the Poll Responses
+"""
+from __future__ import annotations
+
+from sqlalchemy.sql.expression import false
+from sqlalchemy.sql.schema import ForeignKey
+
+from .base_model import BaseModel
+from .db import db
+
+
+class PollResponse(BaseModel):
+    """Definition of the PollResponse entity."""
+
+    __tablename__ = 'poll_responses'
+    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
+    participant_id = db.Column(db.String(255), nullable=False)
+    selected_answer_id = db.Column(db.Integer, ForeignKey('poll_answers.id', ondelete='CASCADE'), nullable=False)
+    poll_id = db.Column(db.Integer, ForeignKey('widget_polls.id', ondelete='CASCADE'), nullable=False)
+    widget_id = db.Column(db.Integer, ForeignKey('widget.id', ondelete='CASCADE'), nullable=False)
+    is_deleted = db.Column(db.Boolean, default=False)
+
+    @classmethod
+    def get_responses(cls, poll_id) -> list[PollResponse]:
+        """Get responses for a poll."""
+        return db.session.query(PollResponse).filter(PollResponse.poll_id == poll_id).all()
+
+    @classmethod
+    def get_responses_by_participant_id(cls, poll_id, participant_id) -> list[PollResponse]:
+        """Get responses for a poll."""
+        return db.session.query(PollResponse).filter(PollResponse.poll_id == poll_id,
+                                                     PollResponse.participant_id == participant_id,
+                                                     PollResponse.is_deleted == false()).all()
+
+    @classmethod
+    def update_response(cls, response_id, response_data: dict) -> PollResponse:
+        """Update a poll response."""
+        response = PollResponse.query.get(response_id)
+        if response:
+            for key, value in response_data.items():
+                setattr(response, key, value)
+            response.save()
+        return response

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

@@ -0,0 +1,65 @@
+"""
+WidgetPoll model class.
+
+Manages the Poll widget
+"""
+from __future__ import annotations
+
+from sqlalchemy import Enum
+from sqlalchemy.sql.schema import ForeignKey
+
+from met_api.models.poll_answers import PollAnswer
+
+from .base_model import BaseModel
+from .db import db
+
+
+class Poll(BaseModel):
+    """Definition of the Poll entity."""
+
+    __tablename__ = 'widget_polls'
+    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
+    title = db.Column(db.String(255), nullable=False)
+    description = db.Column(db.String(2048), nullable=True)
+    status = db.Column(
+        Enum('active', 'inactive', name='poll_status'), default='inactive')
+    widget_id = db.Column(db.Integer, ForeignKey(
+        'widget.id', ondelete='CASCADE'), nullable=False)
+    engagement_id = db.Column(db.Integer, ForeignKey(
+        'engagement.id', ondelete='CASCADE'), nullable=False)
+
+    # Relationship to timeline_event
+    answers = db.relationship(PollAnswer, backref='widget_poll', lazy=True)
+
+    @classmethod
+    def create_poll(cls, widget_id: int, poll_data: dict) -> Poll:
+        """Create a new poll."""
+        poll = cls()
+        poll.widget_id = widget_id
+        poll.title = poll_data.get('title')
+        poll.description = poll_data.get('description')
+        poll.status = poll_data.get('status', 'inactive')
+        poll.engagement_id = poll_data.get('engagement_id')
+        db.session.add(poll)
+        db.session.commit()
+        return poll
+
+    @classmethod
+    def get_polls(cls, widget_id) -> list[Poll]:
+        """Get polls for a widget."""
+        return db.session.query(Poll).filter(Poll.widget_id == widget_id).all()
+
+    @classmethod
+    def update_poll(cls, poll_id, poll_data: dict) -> Poll:
+        """Update a poll and its answers."""
+        poll: Poll = Poll.query.get(poll_id)
+        if poll:
+            # Update poll fields
+            for key in ['title', 'description', 'status', 'widget_id',
+                        'engagement_id']:
+                if key in poll_data:
+                    setattr(poll, key, poll_data[key])
+
+            db.session.commit()
+
+        return poll

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

@@ -50,6 +50,7 @@
 from .engagement_settings import API as ENGAGEMENT_SETTINGS_API
 from .cac_form import API as CAC_FORM_API
 from .widget_timeline import API as WIDGET_TIMELINE_API
+from .widget_poll import API as WIDGET_POLL_API
 
 __all__ = ('API_BLUEPRINT',)
 
@@ -91,3 +92,4 @@
 API.add_namespace(ENGAGEMENT_SETTINGS_API)
 API.add_namespace(CAC_FORM_API, path='/engagements/<int:engagement_id>/cacform')
 API.add_namespace(WIDGET_TIMELINE_API, path='/widgets/<int:widget_id>/timelines')
+API.add_namespace(WIDGET_POLL_API, path='/widgets/<int:widget_id>/polls')