officer end point (#326)

* officer end point

* officer end point

* updated

* update

* clean update

backend/api.py

@@ -11,6 +11,7 @@
 from backend.schemas import spec
 from backend.routes.partners import bp as partners_bp
 from backend.routes.incidents import bp as incidents_bp
+from backend.routes.officers import bp as officers_bp 
 from backend.routes.auth import bp as auth_bp
 from backend.routes.healthcheck import bp as healthcheck_bp
 from backend.utils import dev_only
@@ -130,6 +131,7 @@ def register_routes(app: Flask):
     app.register_blueprint(incidents_bp)
     app.register_blueprint(auth_bp)
     app.register_blueprint(healthcheck_bp)
+    app.register_blueprint(officers_bp)
 
     @app.route("/")
     def hello_world():

backend/routes/officers.py

@@ -0,0 +1,93 @@
+import logging
+from operator import or_
+from typing import Optional
+from venv import logger
+
+from backend.auth.jwt import min_role_required
+from backend.database.models.agency import JURISDICTION
+from backend.mixpanel.mix import track_to_mp
+from mixpanel import MixpanelException
+from backend.database.models.user import UserRole
+from flask import  Blueprint, abort, jsonify, request
+from flask_jwt_extended.view_decorators import jwt_required
+from pydantic import BaseModel
+
+from ..database import  Officer, db, agency_officer, Agency
+from ..schemas import (
+    Officer_orm_to_json,
+    validate,
+)
+
+
+bp = Blueprint("officer_routes", __name__, url_prefix="/api/v1/officers")
+
+class SearchOfficerSchema(BaseModel):
+    officerName: Optional[str] = None
+    location:Optional[str]=None
+    badgeNumber:Optional[str]=None
+    page: Optional[int] = 1
+    perPage: Optional[int] = 20
+
+    class Config:
+        extra = "forbid"
+        schema_extra = {
+            "example": {
+                "officerName": "John Doe",
+                "location":"New York",
+                "badgeNumber":1234,
+                "page": 1,
+                "perPage": 20,
+            }
+        }
+
+@bp.route("/search/officer",methods=["POST"])
+@jwt_required()
+@min_role_required(UserRole.PUBLIC)
+@validate(json=SearchOfficerSchema)
+def search_officer():
+    """Search Officers"""
+    body:SearchOfficerSchema=request.context.json
+
+    query = db.session.query('Officer')
+    try:
+
+        if body.officerName:
+            names = body.officerName.split()
+            first_name = names[0] if len(names) > 0 else ''
+            last_name = names[1] if len(names) > 1 else ''
+            query = Officer.query.filter(or_(
+                Officer.first_name.ilike(f"%{first_name}%"),
+                Officer.last_name.ilike(f"%{last_name}%")
+            ))
+
+        if body.badgeNumber:
+            officer_ids = [result.officer_id for result in db.session.query(agency_officer).filter_by(badge_number=body.badgeNumber).all()]
+            query = Officer.query.filter(Officer.id.in_(officer_ids)).all()
+
+    except Exception as e:
+        abort(422,description=str(e))
+
+    results = query.paginate(
+        page=body.page, per_page=body.perPage, max_per_page=100
+    )
+
+    try:
+        track_to_mp(request, "search_officer", {
+            "officername": body.officerName,
+            "badgeNumber": body.badgeNumber
+        })
+    except MixpanelException as e:
+        logger.error(e)
+    try:
+        return{
+            "results":[
+
+                Officer_orm_to_json(result) for result in results.items
+            ],
+            "page": results.page,
+            "totalPages": results.pages,
+            "totalResults": results.total,
+        }
+    except Exception as e:
+        abort(500, description=str(e))
+

backend/schemas.py

@@ -269,7 +269,33 @@ def incident_orm_to_json(incident: Incident) -> dict[str, Any]:
             "victims",
         },
     )
+# officer end point
 
+_officer_list_attributes=[
+    'first_name',
+    'last_name'
+]
+class _OfficerMixin(BaseModel):
+    @root_validator(pre=True)
+    def none_to_list(cls, values: Dict[str, Any]) -> Dict[str, Any]:
+
+        values = {**values}  # convert mappings to base dict type.
+        for i in _officer_list_attributes:
+            if not values.get(i):
+                values[i] = []
+        return values
+_BaseOfficerSchema = schema_get(Officer)
+
+class OfficerSchema(_BaseOfficerSchema, _OfficerMixin):
+    reported_Officer: Optional[List[_BaseOfficerSchema]]
+
+
+def Officer_orm_to_json(officer: Officer) -> dict:
+    return IncidentSchema.from_orm(officer).dict(
+        exclude_none=True,
+        # Exclude a bunch of currently-unused empty lists
+
+    )
 
 def partner_to_orm(partner: CreatePartnerSchema) -> Partner:
     """Convert the JSON partner into an ORM instance