|
| 1 | +import os |
| 2 | +import sys |
| 3 | +from contextlib import asynccontextmanager |
| 4 | +from typing import Annotated |
| 5 | + |
| 6 | +from fastapi import Header |
| 7 | + |
| 8 | +sys.path.insert(0, os.path.abspath("..")) |
| 9 | + |
| 10 | +from common.app import DetectorBaseAPI as FastAPI |
| 11 | +from detector import Detector |
| 12 | +from scheme import ( |
| 13 | + ContentAnalysisHttpRequest, |
| 14 | + ContentsAnalysisResponse, |
| 15 | + Error, |
| 16 | +) |
| 17 | + |
| 18 | +detector_objects = {} |
| 19 | + |
| 20 | + |
| 21 | +@asynccontextmanager |
| 22 | +async def lifespan(app: FastAPI): |
| 23 | + detector_objects["detector"] = Detector() |
| 24 | + yield |
| 25 | + # Clean up the ML models and release the resources |
| 26 | + detector_objects.clear() |
| 27 | + |
| 28 | + |
| 29 | +app = FastAPI(lifespan=lifespan, dependencies=[]) |
| 30 | + |
| 31 | + |
| 32 | +@app.post( |
| 33 | + "/api/v1/text/contents", |
| 34 | + response_model=ContentsAnalysisResponse, |
| 35 | + description="""Detectors that work on content text, be it user prompt or generated text. \ |
| 36 | + Generally classification type detectors qualify for this. <br>""", |
| 37 | + responses={ |
| 38 | + 404: {"model": Error, "description": "Resource Not Found"}, |
| 39 | + 422: {"model": Error, "description": "Validation Error"}, |
| 40 | + }, |
| 41 | +) |
| 42 | +async def detector_unary_handler( |
| 43 | + request: ContentAnalysisHttpRequest, |
| 44 | + detector_id: Annotated[str, Header(example="en_syntax_slate.38m.hap")], |
| 45 | +): |
| 46 | + return ContentsAnalysisResponse(root=detector_objects["detector"].run(request)) |
0 commit comments