Refactor text input: service layer, repository, reuse validation handler

Author: abhishek-8081

app/api/router.py

@@ -1,7 +1,6 @@
 from fastapi import APIRouter
 
-from app.api.routes import forms, jobs, system, templates, weather, zipcode
-from app.api.routes import input as input_routes
+from app.api.routes import forms, input, jobs, system, templates, weather, zipcode
 from app.core.config import API_PREFIX
 
 api_router = APIRouter()
@@ -11,4 +10,4 @@
 api_router.include_router(jobs.router, prefix=API_PREFIX)
 api_router.include_router(weather.router, prefix=API_PREFIX)
 api_router.include_router(zipcode.router, prefix=API_PREFIX)
-api_router.include_router(input_routes.router, prefix=API_PREFIX)
+api_router.include_router(input.router, prefix=API_PREFIX)

app/api/routes/input.py

@@ -1,65 +1,51 @@
-from datetime import datetime, timezone
 from uuid import UUID
 
 from fastapi import APIRouter, Depends
-from fastapi.responses import JSONResponse
-from sqlmodel import Session, select
+from fastapi.exceptions import RequestValidationError
+from sqlmodel import Session
 
 from app.api.deps import get_db
-from app.api.schemas.enums import InputStatus, InputType
+from app.api.schemas.enums import InputStatus
 from app.api.schemas.input import InputRecordResponse, TextInputRequest, TextInputResponse
 from app.core.config import INPUT_POLL_INTERVAL_SECONDS
 from app.core.errors.base import AppError
-from app.models import Input
+from app.db.repositories import create_input, get_input as repo_get_input
+from app.services.input import InputService
 
 router = APIRouter(prefix="/input", tags=["input"])
 
 
 @router.post("/text", response_model=TextInputResponse, status_code=201)
 def submit_text_input(body: TextInputRequest, db: Session = Depends(get_db)):
-    narrative = body.narrative
-
-    if len(narrative) > 50_000:
+    if len(body.narrative) > 50_000:
         raise AppError(
             "Narrative exceeds maximum length of 50,000 characters",
             status_code=413,
             error_code="NARRATIVE_TOO_LONG",
-            detail={"max_characters": 50_000, "received_characters": len(narrative)},
+            detail={"max_characters": 50_000, "received_characters": len(body.narrative)},
         )
 
-    words = narrative.split()
-    if len(words) < 10:
-        return JSONResponse(
-            status_code=422,
-            content={
-                "error_code": "VALIDATION_ERROR",
-                "message": "Request validation failed",
-                "validation_errors": [
-                    {
-                        "field": "narrative",
-                        "issue": "Must contain at least 10 words",
-                        "value": narrative,
-                    }
-                ],
-            },
+    svc = InputService()
+    try:
+        record = svc.build_text_input(
+            narrative=body.narrative,
+            station_id=body.station_id,
+            responder_badge=body.responder_badge,
+            incident_date_hint=body.incident_date_hint,
+        )
+    except ValueError as exc:
+        raise RequestValidationError(
+            errors=[
+                {
+                    "loc": ("body", "narrative"),
+                    "msg": str(exc),
+                    "input": body.narrative,
+                    "type": "value_error",
+                }
+            ]
         )
 
-    now = datetime.now(timezone.utc)
-    record = Input(
-        input_type=InputType.text,
-        status=InputStatus.ready,
-        transcript=narrative,
-        character_count=len(narrative),
-        word_count=len(words),
-        station_id=body.station_id,
-        responder_badge=body.responder_badge,
-        incident_date_hint=body.incident_date_hint,
-        created_at=now,
-        updated_at=now,
-    )
-    db.add(record)
-    db.commit()
-    db.refresh(record)
+    record = create_input(db, record)
 
     return TextInputResponse(
         input_id=record.input_id,
@@ -73,7 +59,7 @@ def submit_text_input(body: TextInputRequest, db: Session = Depends(get_db)):
 
 @router.get("/{input_id}", response_model=InputRecordResponse)
 def get_input(input_id: UUID, db: Session = Depends(get_db)):
-    record = db.exec(select(Input).where(Input.input_id == input_id)).first()
+    record = repo_get_input(db, input_id)
     if record is None:
         raise AppError(
             f"Input with ID {input_id} not found",

app/api/schemas/input.py

@@ -3,21 +3,19 @@
 from datetime import date, datetime
 from uuid import UUID
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, Field
 
 from app.api.schemas.enums import InputStatus, InputType
 
 
 class TextInputRequest(BaseModel):
-    model_config = ConfigDict()
     narrative: str = Field(min_length=20)
     station_id: str | None = None
     responder_badge: str | None = None
     incident_date_hint: date | None = None
 
 
 class TextInputResponse(BaseModel):
-    model_config = ConfigDict()
     input_id: UUID
     status: InputStatus
     input_type: InputType
@@ -27,7 +25,6 @@ class TextInputResponse(BaseModel):
 
 
 class InputRecordResponse(BaseModel):
-    model_config = ConfigDict()
     input_id: UUID
     input_type: InputType
     status: InputStatus

app/db/repositories.py

@@ -1,5 +1,8 @@
+from uuid import UUID
+
 from sqlmodel import Session, select
-from app.models import Template, FormSubmission, Job
+
+from app.models import Template, FormSubmission, Job, Input
 
 # Templates
 def create_template(session: Session, template: Template) -> Template:
@@ -66,3 +69,15 @@ def delete_form_submission(session: Session, submission: FormSubmission) -> None
     session.delete(submission)
     session.commit()
 
+
+# Inputs
+def create_input(session: Session, input_obj: Input) -> Input:
+    session.add(input_obj)
+    session.commit()
+    session.refresh(input_obj)
+    return input_obj
+
+
+def get_input(session: Session, input_id: UUID) -> Input | None:
+    return session.get(Input, input_id)
+

app/services/input.py

@@ -0,0 +1,30 @@
+from datetime import date, datetime, timezone
+
+from app.api.schemas.enums import InputStatus, InputType
+from app.models import Input
+
+
+class InputService:
+    def build_text_input(
+        self,
+        narrative: str,
+        station_id: str | None = None,
+        responder_badge: str | None = None,
+        incident_date_hint: date | None = None,
+    ) -> Input:
+        words = narrative.split()
+        if len(words) < 10:
+            raise ValueError("Must contain at least 10 words")
+        now = datetime.now(timezone.utc)
+        return Input(
+            input_type=InputType.text,
+            status=InputStatus.ready,
+            transcript=narrative,
+            character_count=len(narrative),
+            word_count=len(words),
+            station_id=station_id,
+            responder_badge=responder_badge,
+            incident_date_hint=incident_date_hint,
+            created_at=now,
+            updated_at=now,
+        )