forked from codeforboston/police-data-trust
-
Notifications
You must be signed in to change notification settings - Fork 0
/
incidents.py
107 lines (88 loc) · 3.03 KB
/
incidents.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
from datetime import datetime
from typing import Optional
from backend.auth.jwt import min_role_required, contributor_has_source
from backend.database.models.user import UserRole
from flask import Blueprint, abort, current_app, request
from flask_jwt_extended.view_decorators import jwt_required
from pydantic import BaseModel
from ..database import Incident, db
from ..schemas import (
CreateIncidentSchema,
incident_orm_to_json,
incident_to_orm,
validate,
)
bp = Blueprint("incident_routes", __name__, url_prefix="/api/v1/incidents")
@bp.route("/get/<int:incident_id>", methods=["GET"])
@jwt_required()
@min_role_required(UserRole.PUBLIC)
@validate()
def get_incidents(incident_id: int):
"""Get a single incident by ID."""
return incident_orm_to_json(Incident.get(incident_id))
@bp.route("/create", methods=["POST"])
@jwt_required()
@min_role_required(UserRole.CONTRIBUTOR)
@contributor_has_source()
@validate(json=CreateIncidentSchema)
def create_incident():
"""Create a single incident.
Cannot be called in production environments
"""
if current_app.env == "production":
abort(418)
try:
incident = incident_to_orm(request.context.json)
except Exception:
abort(400)
created = incident.create()
return incident_orm_to_json(created)
class SearchIncidentsSchema(BaseModel):
location: Optional[str] = None
startTime: Optional[datetime] = None
endTime: Optional[datetime] = None
description: Optional[str] = None
page: Optional[int] = 1
perPage: Optional[int] = 20
class Config:
extra = "forbid"
schema_extra = {
"example": {
"description": "Test description",
"endTime": "2019-12-01 00:00:00",
"location": "Location 1",
"startTime": "2019-09-01 00:00:00",
}
}
@bp.route("/search", methods=["POST"])
@jwt_required()
@min_role_required(UserRole.PUBLIC)
@validate(json=SearchIncidentsSchema)
def search_incidents():
"""Search Incidents."""
body: SearchIncidentsSchema = request.context.json
query = db.session.query(Incident)
if body.location:
# TODO: Replace with .match, which uses `@@ to_tsquery` for full-text
# search
#
# TODO: eventually replace with geosearch. Geocode records and integrate
# PostGIS
query = query.filter(Incident.location.ilike(f"%{body.location}%"))
if body.startTime:
query = query.filter(Incident.time_of_incident >= body.startTime)
if body.endTime:
query = query.filter(Incident.time_of_incident <= body.endTime)
if body.description:
query = query.filter(
Incident.description.ilike(f"%{body.description}%")
)
results = query.paginate(
page=body.page, per_page=body.perPage, max_per_page=100
)
return {
"results": [incident_orm_to_json(result) for result in results.items],
"page": results.page,
"totalPages": results.pages,
"totalResults": results.total,
}