Appendix backend

dinesh-aot · dinesh-aot · commit 366ef2cdbac5 · 2025-02-26T12:51:14.000-08:00
diff --git a/compliance-api/migrations/versions/6e1dd95e1a4d_appendix_fkey_inspection_added.py b/compliance-api/migrations/versions/6e1dd95e1a4d_appendix_fkey_inspection_added.py
@@ -0,0 +1,47 @@
+"""appendix-fkey-inspection added
+
+
+Revision ID: 6e1dd95e1a4d
+Revises: e8a29d9b9a0d
+Create Date: 2025-02-26 11:43:42.768435
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '6e1dd95e1a4d'
+down_revision = 'e8a29d9b9a0d'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('appendices', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('inspection_id', sa.Integer(), nullable=False, comment='The unique identifier of the inspection'))
+        batch_op.drop_constraint('appendices_appendix_no_key', type_='unique')
+        batch_op.create_index('unique_non_deleted_appendix_number', ['inspection_id', 'appendix_no'], unique=True, postgresql_where=sa.text('is_deleted = false'))
+        batch_op.create_foreign_key('appendix_inspection_id_inspection_id_fkey', 'inspections', ['inspection_id'], ['id'])
+
+    with op.batch_alter_table('appendices_version', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('inspection_id', sa.Integer(), autoincrement=False, nullable=True, comment='The unique identifier of the inspection'))
+        batch_op.add_column(sa.Column('inspection_id_mod', sa.Boolean(), server_default=sa.text('false'), nullable=False))
+
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('appendices_version', schema=None) as batch_op:
+        batch_op.drop_column('inspection_id_mod')
+        batch_op.drop_column('inspection_id')
+
+    with op.batch_alter_table('appendices', schema=None) as batch_op:
+        batch_op.drop_constraint('appendix_inspection_id_inspection_id_fkey', type_='foreignkey')
+        batch_op.drop_index('unique_non_deleted_appendix_number', postgresql_where=sa.text('is_deleted = false'))
+        batch_op.create_unique_constraint('appendices_appendix_no_key', ['appendix_no'])
+        batch_op.drop_column('inspection_id')
+
+    # ### end Alembic commands ###
diff --git a/compliance-api/src/compliance_api/models/appendix.py b/compliance-api/src/compliance_api/models/appendix.py
@@ -1,6 +1,7 @@
 """Model to handle appendices."""
 
-from sqlalchemy import Column, Integer, String
+from sqlalchemy import Boolean, Column, ForeignKey, Index, Integer, String
+from sqlalchemy.orm import relationship
 
 from .base_model import BaseModelVersioned
 
@@ -11,5 +12,36 @@ class Appendix(BaseModelVersioned):
     __tablename__ = "appendices"
 
     id = Column(Integer, primary_key=True, autoincrement=True)
-    appendix_no = Column(Integer, unique=True, nullable=False)
+    inspection_id = Column(
+        Integer,
+        ForeignKey("inspections.id", name="appendix_inspection_id_inspection_id_fkey"),
+        nullable=False,
+        comment="The unique identifier of the inspection",
+    )
+    appendix_no = Column(Integer, nullable=False)
     document_title = Column(String, nullable=False)
+    is_deleted = Column(Boolean, default=False, server_default="f", nullable=False)
+    inspection = relationship("Inspection", foreign_keys=[inspection_id], lazy="select")
+    __table_args__ = (
+        Index(
+            "unique_non_deleted_appendix_number",  # Index name
+            "inspection_id",
+            "appendix_no",
+            unique=True,
+            postgresql_where=(is_deleted is False),  # Condition for uniqueness
+        ),
+    )
+
+    @classmethod
+    def get_by_no_nd_inspection(cls, appendix_no: int, inspection_id: int):
+        """Get appendix by name."""
+        return cls.query.filter_by(
+            appendix_no=appendix_no, inspection_id=inspection_id, is_deleted=False
+        ).first()
+
+    @classmethod
+    def get_by_inspection_id(cls, inspection_id: int):
+        """Get all appendices by inspection id."""
+        return cls.query.filter_by(
+            inspection_id=inspection_id, is_deleted=False, is_active=True
+        ).all()
diff --git a/compliance-api/src/compliance_api/resources/__init__.py b/compliance-api/src/compliance_api/resources/__init__.py
@@ -25,6 +25,7 @@
 
 from .agency import API as AGENCY_API
 from .apihelper import Api
+from .appendix import API as APPENDIX_API
 from .case_file import API as CASE_FILE_API
 from .complaint import API as COMPLAINT_API
 from .compliance_finding import API as COMPLIANCE_FINDING_API
@@ -93,3 +94,4 @@
 )
 API.add_namespace(DOCUMENT_TYPE_API)
 API.add_namespace(REQUIREMENT_TYPE_API)
+API.add_namespace(APPENDIX_API)
diff --git a/compliance-api/src/compliance_api/resources/appendix.py b/compliance-api/src/compliance_api/resources/appendix.py
@@ -0,0 +1,128 @@
+# Copyright © 2024 Province of British Columbia
+#
+# Licensed under the Apache License, Version 2.0 (the 'License');
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an 'AS IS' BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""API endpoints for managing Appendix resource."""
+
+from http import HTTPStatus
+
+from flask import request
+from flask_restx import Namespace, Resource
+
+from compliance_api.auth import auth
+from compliance_api.exceptions import ResourceNotFoundError
+from compliance_api.schemas import AppendixCreateSchema, AppendixSchema
+from compliance_api.services import AppendixService
+from compliance_api.utils.enum import PermissionEnum
+from compliance_api.utils.util import cors_preflight
+
+from .apihelper import Api as ApiHelper
+
+
+API = Namespace("appendices", description="Endpoints for Appendix Management")
+
+appendix_request_model = ApiHelper.convert_ma_schema_to_restx_model(
+    API, AppendixCreateSchema(), "Appendix"
+)
+appendix_list_model = ApiHelper.convert_ma_schema_to_restx_model(
+    API, AppendixSchema(), "AppendixList"
+)
+
+
+@cors_preflight("GET, OPTIONS, POST")
+@API.route("", methods=["POST", "GET", "OPTIONS"])
+class Appendices(Resource):
+    """Resource for managing appendices."""
+
+    @staticmethod
+    @API.response(code=200, description="Success", model=[appendix_list_model])
+    @API.doc(
+        params={
+            "inspection_id": {
+                "description": "The unique identifier of the inspection",
+                "type": "integer",
+                "required": False,
+            }
+        }
+    )
+    @ApiHelper.swagger_decorators(API, endpoint_description="Fetch all appendices")
+    @auth.require
+    def get():
+        """Fetch all appendices."""
+        inspection_id = request.args.get("inspection_id", None)
+        if inspection_id:
+            appendices = AppendixService.get_by_inspection_id(inspection_id)
+        else:
+            appendices = AppendixService.get_all()
+        appendix_list_schema = AppendixSchema(many=True)
+        return appendix_list_schema.dump(appendices), HTTPStatus.OK
+
+    @staticmethod
+    @auth.require
+    @ApiHelper.swagger_decorators(API, endpoint_description="Create an Appendix")
+    @API.expect(appendix_request_model)
+    @API.response(code=201, model=appendix_list_model, description="AppendixCreated")
+    @API.response(400, "Bad Request")
+    @auth.has_one_of_roles([PermissionEnum.SUPERUSER, PermissionEnum.ADMIN])
+    def post():
+        """Create a Appendix."""
+        appendix_data = AppendixCreateSchema().load(API.payload)
+        created_appendix = AppendixService.create(appendix_data)
+        return AppendixSchema().dump(created_appendix), HTTPStatus.CREATED
+
+
+@cors_preflight("GET, OPTIONS, PATCH, DELETE")
+@API.route("/<int:appendix_id>", methods=["PATCH", "GET", "OPTIONS", "DELETE"])
+@API.doc(params={"appendix_id": "The unique identifier of appendix"})
+class Appendix(Resource):
+    """Resource for managing a single Appendix."""
+
+    @staticmethod
+    @auth.require
+    @ApiHelper.swagger_decorators(API, endpoint_description="Fetch an appendix by id")
+    @API.response(code=200, model=appendix_list_model, description="Success")
+    @API.response(404, "Not Found")
+    def get(appendix_id):
+        """Fetch an appendix by id."""
+        appendix = AppendixService.get_by_id(appendix_id)
+        if not appendix:
+            raise ResourceNotFoundError(f"Appendix with {appendix_id} not found")
+        return AppendixSchema().dump(Appendix), HTTPStatus.OK
+
+    @staticmethod
+    @auth.require
+    @ApiHelper.swagger_decorators(API, endpoint_description="Update an appendix by id")
+    @API.expect(appendix_request_model)
+    @API.response(code=200, model=appendix_list_model, description="Success")
+    @API.response(400, "Bad Request")
+    @API.response(404, "Not Found")
+    @auth.has_one_of_roles([PermissionEnum.SUPERUSER, PermissionEnum.ADMIN])
+    def patch(appendix_id):
+        """Update an Appendix by id."""
+        appendix_data = AppendixCreateSchema().load(API.payload)
+        updated_appendix = AppendixService.update(appendix_id, appendix_data)
+        if not updated_appendix:
+            raise ResourceNotFoundError(f"Appendix with {appendix_id} not found")
+        return AppendixSchema().dump(updated_appendix), HTTPStatus.OK
+
+    @staticmethod
+    @auth.require
+    @ApiHelper.swagger_decorators(API, endpoint_description="Delete an appendix by id")
+    @API.response(code=200, model=appendix_list_model, description="Deleted")
+    @API.response(404, "Not Found")
+    @auth.has_one_of_roles([PermissionEnum.SUPERUSER, PermissionEnum.ADMIN])
+    def delete(appendix_id):
+        """Delete an appendix by id."""
+        deleted_appendix = AppendixService.delete(appendix_id)
+        if not deleted_appendix:
+            raise ResourceNotFoundError(f"Appendix with {appendix_id} not found")
+        return AppendixSchema().dump(deleted_appendix), HTTPStatus.OK
diff --git a/compliance-api/src/compliance_api/schemas/__init__.py b/compliance-api/src/compliance_api/schemas/__init__.py
@@ -13,6 +13,7 @@
 # limitations under the License.
 """Exposes all of the schemas in the compliance_api."""
 from .agency import AgencyCreateSchema, AgencySchema
+from .appendix import AppendixCreateSchema, AppendixSchema
 from .case_file import (
     CaseFileCreateSchema, CaseFileLinkCreateSchema, CaseFileLinkSchema, CaseFileOfficerSchema, CaseFileOptionSchema,
     CaseFileSchema, CaseFileStatusSchema, CaseFileUnlinkSchema, CaseFileUpdateSchema)
diff --git a/compliance-api/src/compliance_api/schemas/appendix.py b/compliance-api/src/compliance_api/schemas/appendix.py
@@ -0,0 +1,47 @@
+# Copyright © 2024 Province of British Columbia
+#
+# Licensed under the Apache License, Version 2.0 (the 'License');
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an 'AS IS' BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Appendix Schema."""
+from marshmallow import EXCLUDE, fields
+
+from compliance_api.models.appendix import Appendix
+
+from .base_schema import AutoSchemaBase, BaseSchema
+
+
+class AppendixSchema(AutoSchemaBase):  # pylint: disable=too-many-ancestors
+    """Appendix schema."""
+
+    class Meta(AutoSchemaBase.Meta):  # pylint: disable=too-few-public-methods
+        """Exclude unknown fields in the deserialized output."""
+
+        unknown = EXCLUDE
+        model = Appendix
+        include_fk = True
+
+
+class AppendixCreateSchema(BaseSchema):  # pylint: disable=too-many-ancestors
+    """Appendix create Schema."""
+
+    appendix_no = fields.Integer(
+        metadata={"description": "The unique number of appendix"},
+        required=True,
+    )
+    inspection_id = fields.Integer(
+        metadata={"description": "The inspection id"},
+        required=True
+    )
+    document_title = fields.Str(
+        metadata={"description": "The title of the document"},
+        required=True
+    )
diff --git a/compliance-api/src/compliance_api/services/__init__.py b/compliance-api/src/compliance_api/services/__init__.py
@@ -13,6 +13,7 @@
 # limitations under the License.
 """Exposes all of the Services used in the compliance_api."""
 from .agency import AgencyService
+from .appendix import AppendixService
 from .case_file import CaseFileService
 from .complaint import ComplaintService
 from .compliance_finding import ComplianceFindingService
diff --git a/compliance-api/src/compliance_api/services/appendix.py b/compliance-api/src/compliance_api/services/appendix.py
@@ -0,0 +1,93 @@
+"""Service for appendix management."""
+
+from compliance_api.exceptions import ResourceExistsError, ResourceNotFoundError
+from compliance_api.models import db
+from compliance_api.models.appendix import Appendix as AppendixModel
+from compliance_api.models.inspection import Inspection as InspectionModel
+
+
+class AppendixService:
+    """Appendix management service."""
+
+    @classmethod
+    def get_by_id(cls, appendix_id):
+        """Get appendix by id."""
+        appendix = AppendixModel.find_by_id(appendix_id)
+        return appendix
+
+    @classmethod
+    def get_by_inspection_id(cls, inspection_id):
+        """Get appendix by id."""
+        appendices = AppendixModel.get_by_inspection_id(inspection_id)
+        return appendices
+
+    @classmethod
+    def get_all(cls):
+        """Get all appendices."""
+        appendices = AppendixModel.get_all(default_filters=False)
+        return appendices
+
+    @classmethod
+    def create(cls, appendix_data: dict, commit=True):
+        """Create appendix."""
+        inspection_id = appendix_data.get("inspection_id")
+        _check_existence_by_no(appendix_data.get("appendix_no"), inspection_id, None)
+        _inspection_check(inspection_id)
+        appendix = AppendixModel(**appendix_data)
+        appendix.flush()
+        if commit:
+            db.session.commit()
+        return appendix
+
+    @classmethod
+    def update(cls, appendix_id, appendix_data, commit=True):
+        """Update appendix."""
+        inspection_id = appendix_data.get("inspection_id")
+        _check_existence_by_no(
+            appendix_data.get("appendix_no"),
+            inspection_id,
+            appendix_id,
+        )
+        _inspection_check(appendix_data.get("inspection_id"))
+        appendix = AppendixModel.find_by_id(appendix_id)
+        if not appendix:
+            return None
+        appendix.update(appendix_data, commit=False)
+        db.session.flush()
+        if commit:
+            db.session.commit()
+        return appendix
+
+    @classmethod
+    def delete(cls, agency_id, commit=True):
+        """Delete the appendix entity permenantly from database."""
+        appendix = AppendixModel.find_by_id(agency_id)
+        if not appendix:
+            return None
+        appendix.is_deleted = True
+        appendix.is_active = False
+        db.session.flush()
+        if commit:
+            db.session.commit()
+        return appendix
+
+
+def _check_existence_by_no(
+    appendix_no: str, inspection_id: int, appendix_id: int = None
+):
+    """Check if the appendix exists."""
+    existing_appendix = AppendixModel.get_by_no_nd_inspection(
+        appendix_no, inspection_id
+    )
+    if existing_appendix and (not appendix_id or existing_appendix.id != appendix_id):
+        raise ResourceExistsError(f"Appendix with the number {appendix_no} exists")
+
+
+def _inspection_check(inspection_id):
+    """Check if the inspection and requirement exists."""
+    inspection = InspectionModel.find_by_id(inspection_id)
+    if not inspection:
+        raise ResourceNotFoundError(
+            f"Inspection with given ID {inspection_id} not found"
+        )
+    return inspection
