From 443e786241c797e244c9294e42ce040e04ad261a Mon Sep 17 00:00:00 2001 From: Darrell Malone Jr Date: Sun, 29 Sep 2024 23:12:34 -0500 Subject: [PATCH] Add Update to routes - officers - agencies - partners Fix temporary pydantic schema for agencies --- backend/routes/agencies.py | 51 +++++---- backend/routes/officers.py | 23 ++-- backend/routes/partners.py | 48 +++++++-- backend/routes/tmp/pydantic/agencies.py | 133 ++++++++++++++++++++++++ 4 files changed, 209 insertions(+), 46 deletions(-) create mode 100644 backend/routes/tmp/pydantic/agencies.py diff --git a/backend/routes/agencies.py b/backend/routes/agencies.py index 8ca167af..7f1d4941 100644 --- a/backend/routes/agencies.py +++ b/backend/routes/agencies.py @@ -3,9 +3,13 @@ from operator import and_ from typing import Optional, List from backend.auth.jwt import min_role_required +from backend.schemas import ( + validate_request, paginate_results, ordered_jsonify, + NodeConflictException) from backend.mixpanel.mix import track_to_mp from backend.database.models.user import UserRole from backend.database.models.agency import Agency +from .tmp.pydantic.agencies import CreateAgency, UpdateAgency from flask import Blueprint, abort, request from flask_jwt_extended.view_decorators import jwt_required from pydantic import BaseModel @@ -33,16 +37,19 @@ class AddOfficerListSchema(BaseModel): @bp.route("/", methods=["POST"]) @jwt_required() @min_role_required(UserRole.CONTRIBUTOR) -#@validate(json=CreateAgencySchema) +@validate_request(CreateAgency) def create_agency(): logger = logging.getLogger("create_agency") """Create an agency profile. User must be a Contributor to create an agency. Must include a name and jurisdiction. """ + body: CreateAgency = request.validated_body try: - agency = Agency.from_dict(request.context.json) + agency = Agency.from_dict(body.dict()) + except NodeConflictException: + abort(409, description="Agency already exists") except Exception as e: logger.error(f"Error, Agency.from_dict: {e}") abort(400) @@ -58,13 +65,13 @@ def create_agency(): # Get agency profile -@bp.route("/", methods=["GET"]) +@bp.route("/", methods=["GET"]) @jwt_required() @min_role_required(UserRole.PUBLIC) -# @validate() -def get_agency(agency_id: int): +def get_agency(agency_id: str): """Get an agency profile. """ + logger = logging.getLogger("get_agency") agency = Agency.nodes.get_or_none(uid=agency_id) if agency is None: abort(404, description="Agency not found") @@ -75,19 +82,22 @@ def get_agency(agency_id: int): # Update agency profile -@bp.route("/", methods=["PUT"]) +@bp.route("/", methods=["PUT"]) @jwt_required() @min_role_required(UserRole.CONTRIBUTOR) -# @validate() -def update_agency(agency_id: int): +@validate_request(UpdateAgency) +def update_agency(agency_uid: str): """Update an agency profile. """ - agency = Agency.nodes.get_or_none(uid=agency_id) + logger = logging.getLogger("update_agency") + body: UpdateAgency = request.validated_body + agency = Agency.nodes.get_or_none(uid=agency_uid) if agency is None: abort(404, description="Agency not found") try: - agency = Agency.from_dict(request.context.json) + agency = Agency.from_dict(body.dict(), agency_uid) + agency.refresh() track_to_mp( request, "update_agency", @@ -101,11 +111,10 @@ def update_agency(agency_id: int): # Delete agency profile -@bp.route("/", methods=["DELETE"]) +@bp.route("/", methods=["DELETE"]) @jwt_required() @min_role_required(UserRole.ADMIN) -# @validate() -def delete_agency(agency_id: int): +def delete_agency(agency_id: str): """Delete an agency profile. Must be an admin to delete an agency. """ @@ -131,7 +140,6 @@ def delete_agency(agency_id: int): @bp.route("/", methods=["GET"]) @jwt_required() @min_role_required(UserRole.PUBLIC) -# @validate() def get_all_agencies(): """Get all agencies. Accepts Query Parameters for pagination: @@ -143,20 +151,9 @@ def get_all_agencies(): q_per_page = args.get("per_page", 20, type=int) all_agencies = Agency.nodes.all() - pagination = all_agencies.paginate( - page=q_page, per_page=q_per_page, max_per_page=100 - ) + results = paginate_results(all_agencies, q_page, q_per_page) - try: - return { - "results": [ - agency.to_json() for agency in pagination.items], - "page": pagination.page, - "totalPages": pagination.pages, - "totalResults": pagination.total, - } - except Exception as e: - abort(500, description=str(e)) + return ordered_jsonify(results), 200 # # Add officer employment information diff --git a/backend/routes/officers.py b/backend/routes/officers.py index 5dc80b8f..6848bddc 100644 --- a/backend/routes/officers.py +++ b/backend/routes/officers.py @@ -157,13 +157,13 @@ def create_officer(): # Get an officer profile -@bp.route("/", methods=["GET"]) +@bp.route("/", methods=["GET"]) @jwt_required() @min_role_required(UserRole.PUBLIC) -def get_officer(officer_id: int): +def get_officer(officer_uid: int): """Get an officer profile. """ - o = Officer.nodes.get_or_none(uid=officer_id) + o = Officer.nodes.get_or_none(uid=officer_uid) if o is None: abort(404, description="Officer not found") return o.to_json() @@ -190,19 +190,21 @@ def get_all_officers(): # Update an officer profile -@bp.route("/", methods=["PUT"]) +@bp.route("/", methods=["PUT"]) @jwt_required() @min_role_required(UserRole.CONTRIBUTOR) @validate_request(UpdateOfficer) -def update_officer(officer_id: int): +def update_officer(officer_uid: str): """Update an officer profile. """ - o = Officer.nodes.get_or_none(uid=officer_id) + body: UpdateOfficer = request.validated_body + o = Officer.nodes.get_or_none(uid=officer_uid) if o is None: abort(404, description="Officer not found") try: - o = Officer.from_dict(request.context.json) + o = Officer.from_dict(body.dict(), officer_uid) + o.refresh() except Exception as e: abort(400, description=str(e)) @@ -217,15 +219,14 @@ def update_officer(officer_id: int): # Delete an officer profile -@bp.route("/", methods=["DELETE"]) +@bp.route("/", methods=["DELETE"]) @jwt_required() @min_role_required(UserRole.ADMIN) -# @validate() -def delete_officer(officer_id: int): +def delete_officer(officer_uid: str): """Delete an officer profile. Must be an admin to delete an officer. """ - o = Officer.nodes.get_or_none(uid=officer_id) + o = Officer.nodes.get_or_none(uid=officer_uid) if o is None: abort(404, description="Officer not found") try: diff --git a/backend/routes/partners.py b/backend/routes/partners.py index 9cb823af..ba644528 100644 --- a/backend/routes/partners.py +++ b/backend/routes/partners.py @@ -4,8 +4,10 @@ from backend.auth.jwt import min_role_required from backend.mixpanel.mix import track_to_mp from backend.database.models.user import User, UserRole -from ..schemas import validate_request, paginate_results, ordered_jsonify -from .tmp.pydantic.partners import CreatePartner +from ..schemas import ( + validate_request, paginate_results, ordered_jsonify, + NodeConflictException) +from .tmp.pydantic.partners import CreatePartner, UpdatePartner from flask import Blueprint, abort, current_app, request, jsonify from flask_jwt_extended import get_jwt from flask_jwt_extended.view_decorators import jwt_required @@ -26,12 +28,12 @@ bp = Blueprint("partner_routes", __name__, url_prefix="/api/v1/partners") -@bp.route("/", methods=["GET"]) +@bp.route("/", methods=["GET"]) @jwt_required() @min_role_required(UserRole.PUBLIC) -def get_partners(partner_id: str): - """Get a single partner by ID.""" - p = Partner.nodes.get_or_none(uid=partner_id) +def get_partners(partner_uid: str): + """Get a single partner by UID.""" + p = Partner.nodes.get_or_none(uid=partner_uid) if p is None: abort(404, description="Partner not found") return p.to_json() @@ -60,6 +62,8 @@ def create_partner(): # Creates a new instance of the Partner and saves it to the DB try: new_p = Partner.from_dict(body.dict()) + except NodeConflictException: + abort(409, description="Partner already exists") except Exception as e: abort( 400, @@ -72,8 +76,9 @@ def create_partner(): } ) # update to UserRole contributor status - current_user.role = UserRole.CONTRIBUTOR.value - current_user.save() + if current_user.role_enum.get_value() < UserRole.CONTRIBUTOR.get_value(): + current_user.role = UserRole.CONTRIBUTOR.value + current_user.save() logger.info(f"User {current_user.uid} created partner {new_p.name}") track_to_mp(request, "create_partner", { @@ -108,6 +113,33 @@ def get_all_partners(): return ordered_jsonify(results), 200 +@bp.route("/", methods=["PATCH"]) +@jwt_required() +@min_role_required(UserRole.PUBLIC) +@validate_request(UpdatePartner) +def update_partner(partner_uid: str): + """Update a partner's information.""" + body: UpdatePartner = request.validated_body + current_user = User.get(get_jwt()["sub"]) + p = Partner.nodes.get_or_none(uid=partner_uid) + if p is None: + abort(404, description="Partner not found") + + if p.members.is_connected(current_user): + rel = p.members.relationship(current_user) + if not rel.is_administrator(): + abort(403, description="Not authorized to update partner") + else: + abort(403, description="Not authorized to update partner") + + try: + p.from_dict(body.dict(), partner_uid) + p.refresh() + return p.to_json() + except Exception as e: + abort(400, description=str(e)) + + @bp.route("//members/", methods=["GET"]) @jwt_required() @min_role_required(UserRole.PUBLIC) diff --git a/backend/routes/tmp/pydantic/agencies.py b/backend/routes/tmp/pydantic/agencies.py new file mode 100644 index 00000000..d5b8efa2 --- /dev/null +++ b/backend/routes/tmp/pydantic/agencies.py @@ -0,0 +1,133 @@ +from pydantic import BaseModel, Field +from typing import List, Optional, Dict, Any, Union +from .common import PaginatedResponse + + +class BaseAgency(BaseModel): + name: Optional[str] = Field(None, description="Name of the agency") + hq_address: Optional[str] = Field(None, description="Address of the agency") + hq_city: Optional[str] = Field(None, description="City of the agency") + hq_state: Optional[str] = Field(None, description="State of the agency") + hq_zip: Optional[str] = Field(None, description="Zip code of the agency") + jurisdiction: Optional[str] = Field(None, description="Jurisdiction of the agency") + phone: Optional[str] = Field(None, description="Phone number of the agency") + email: Optional[str] = Field(None, description="Email of the agency") + website_url: Optional[str] = Field(None, description="Website of the agency") + + +class CreateAgency(BaseAgency, BaseModel): + name: Optional[str] = Field(None, description="Name of the agency") + hq_address: Optional[str] = Field(None, description="Address of the agency") + hq_city: Optional[str] = Field(None, description="City of the agency") + hq_state: Optional[str] = Field(None, description="State of the agency") + hq_zip: Optional[str] = Field(None, description="Zip code of the agency") + jurisdiction: Optional[str] = Field(None, description="Jurisdiction of the agency") + phone: Optional[str] = Field(None, description="Phone number of the agency") + email: Optional[str] = Field(None, description="Email of the agency") + website_url: Optional[str] = Field(None, description="Website of the agency") + + +class UpdateAgency(BaseAgency, BaseModel): + name: Optional[str] = Field(None, description="Name of the agency") + hq_address: Optional[str] = Field(None, description="Address of the agency") + hq_city: Optional[str] = Field(None, description="City of the agency") + hq_state: Optional[str] = Field(None, description="State of the agency") + hq_zip: Optional[str] = Field(None, description="Zip code of the agency") + jurisdiction: Optional[str] = Field(None, description="Jurisdiction of the agency") + phone: Optional[str] = Field(None, description="Phone number of the agency") + email: Optional[str] = Field(None, description="Email of the agency") + website_url: Optional[str] = Field(None, description="Website of the agency") + + +class Agency(BaseAgency, BaseModel): + name: Optional[str] = Field(None, description="Name of the agency") + hq_address: Optional[str] = Field(None, description="Address of the agency") + hq_city: Optional[str] = Field(None, description="City of the agency") + hq_state: Optional[str] = Field(None, description="State of the agency") + hq_zip: Optional[str] = Field(None, description="Zip code of the agency") + jurisdiction: Optional[str] = Field(None, description="Jurisdiction of the agency") + phone: Optional[str] = Field(None, description="Phone number of the agency") + email: Optional[str] = Field(None, description="Email of the agency") + website_url: Optional[str] = Field(None, description="Website of the agency") + uid: Optional[str] = Field(None, description="Unique identifier for the agency") + officers_url: Optional[str] = Field(None, description="URL to get a list of officers for this agency") + units_url: Optional[str] = Field(None, description="URL to get a list of units for this agency") + + +class AgencyList(PaginatedResponse, BaseModel): + results: Optional[List[Agency]] = None + + +class BaseUnit(BaseModel): + """Base properties for a unit""" + name: Optional[str] = Field(None, description="Name of the unit") + website_url: Optional[str] = Field(None, description="Website of the unit") + phone: Optional[str] = Field(None, description="Phone number of the unit") + email: Optional[str] = Field(None, description="Email of the unit") + description: Optional[str] = Field(None, description="Description of the unit") + address: Optional[str] = Field(None, description="Street address of the unit") + zip: Optional[str] = Field(None, description="Zip code of the unit") + date_established: Optional[str] = Field(None, description="The date that this unit was established by its parent agency.") + + +class CreateUnit(BaseUnit, BaseModel): + name: str = Field(..., description="Name of the unit") + website_url: Optional[str] = Field(None, description="Website of the unit") + phone: Optional[str] = Field(None, description="Phone number of the unit") + email: Optional[str] = Field(None, description="Email of the unit") + description: Optional[str] = Field(None, description="Description of the unit") + address: Optional[str] = Field(None, description="Street address of the unit") + zip: Optional[str] = Field(None, description="Zip code of the unit") + date_established: Optional[str] = Field(None, description="The date that this unit was established by its parent agency.") + commander_uid: Optional[str] = Field(None, description="The UID of the unit's current commander.") + + +class UpdateUnit(BaseUnit, BaseModel): + name: Optional[str] = Field(None, description="Name of the unit") + website_url: Optional[str] = Field(None, description="Website of the unit") + phone: Optional[str] = Field(None, description="Phone number of the unit") + email: Optional[str] = Field(None, description="Email of the unit") + description: Optional[str] = Field(None, description="Description of the unit") + address: Optional[str] = Field(None, description="Street address of the unit") + zip: Optional[str] = Field(None, description="Zip code of the unit") + date_established: Optional[str] = Field(None, description="The date that this unit was established by its parent agency.") + commander_uid: Optional[str] = Field(None, description="The UID of the unit's current commander.") + + +class Unit(BaseUnit, BaseModel): + name: Optional[str] = Field(None, description="Name of the unit") + website_url: Optional[str] = Field(None, description="Website of the unit") + phone: Optional[str] = Field(None, description="Phone number of the unit") + email: Optional[str] = Field(None, description="Email of the unit") + description: Optional[str] = Field(None, description="Description of the unit") + address: Optional[str] = Field(None, description="Street address of the unit") + zip: Optional[str] = Field(None, description="Zip code of the unit") + date_established: Optional[str] = Field(None, description="The date that this unit was established by its parent agency.") + uid: Optional[str] = Field(None, description="Unique identifier for the unit") + commander_history_url: Optional[str] = Field(None, description="-| URL that returns the past commanders of the unit and the period of their respective commands.") + agency_url: Optional[str] = Field(None, description="URL to get the agency that this unit belongs to.") + officers_url: Optional[str] = Field(None, description="URL to get a list of officers for this unit.") + + +class UnitList(PaginatedResponse, BaseModel): + results: Optional[List[Unit]] = None + + +class AddOfficer(BaseModel): + officer_uid: str = Field(..., description="The uid of the officer") + earliest_employment: Optional[str] = Field(None, description="The earliest date of employment") + latest_employment: Optional[str] = Field(None, description="The latest date of employment") + badge_number: str = Field(..., description="The badge number of the officer") + unit_uid: str = Field(..., description="The UID of the unit the officer is assigned to.") + highest_rank: Optional[str] = Field(None, description="The highest rank the officer has held during their employment.") + commander: Optional[bool] = Field(None, description="-| If true, this officer will be added as the commander of the unit for the specified time period.") + + +class AddOfficerList(BaseModel): + officers: List[AddOfficer] = ... + + +class AddOfficerFailed(BaseModel): + officer_uid: Optional[str] = Field(None, description="The uid of the officer") + reason: Optional[str] = Field(None, description="The reason the employment record could not be added") +