add IngestService to context manager

src/nv_ingest/api/dependency.py

@@ -0,0 +1,36 @@
+# SPDX-FileCopyrightText: Copyright (c) 2024-25, NVIDIA CORPORATION & AFFILIATES.
+# All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+import logging
+from fastapi import Request, HTTPException
+from nv_ingest.framework.util.service.meta.ingest.ingest_service_meta import IngestServiceMeta
+# The only place that knows the concrete class:
+from nv_ingest.framework.util.service.impl.ingest.redis_ingest_service import RedisIngestService
+
+logger = logging.getLogger(__name__)
+
+def create_ingest_service() -> IngestServiceMeta:
+    """
+    Gather the appropriate Ingestion Service to use for the nv-ingest endpoint.
+    Construct the concrete ingest service (Redis today).
+    Keeping this import/choice here isolates main.py and routers from the impl.
+    In the future, you can swap this to read env or return a different impl.
+    """
+
+    logger.debug("Creating RedisIngestService singleton for dependency injection")
+    return RedisIngestService.get_instance()
+
+async def get_ingest_service(request: Request) -> IngestServiceMeta:
+    """
+    FastAPI dependency used by routers. Prefer the lifespan instance on app.state.
+    As a safe fallback (e.g., unit tests without app startup), lazily create one.
+    """
+    svc = getattr(request.app.state, "ingest_service", None)
+    if svc is None:
+        # Fallback keeps local tests/dev working without lifespan.
+        # If you prefer to enforce startup, raise 503 instead of creating.
+        logger.warning("ingest_service not found on app.state; creating a local instance")
+        svc = create_ingest_service()
+        # do NOT stash it on app.state here to avoid surprising test sharing
+    return svc

src/nv_ingest/api/main.py

@@ -15,9 +15,29 @@
 from .v1.health import router as HealthApiRouter
 from .v1.ingest import router as IngestApiRouter
 from .v1.metrics import router as MetricsApiRouter
+from .dependency import create_ingest_service
 
 logger = logging.getLogger(__name__)
 
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # Build once per process; app owns the instance
+    app.state.ingest_service = create_ingest_service()
+    try:
+        yield
+    finally:
+        svc = getattr(app.state, "ingest_service", None)
+        if svc is not None:
+            close = getattr(svc, "close", None)
+            if callable(close):
+                try:
+                    maybe = close()
+                    import asyncio
+                    if asyncio.iscoroutine(maybe):
+                        await maybe
+                except Exception:
+                    logger.exception("Error closing ingest service during shutdown")
+
 # nv-ingest FastAPI app declaration
 app = FastAPI(
     title="NV-Ingest Microservice",

src/nv_ingest/api/v1/ingest.py

@@ -33,22 +33,15 @@
 from nv_ingest_client.primitives.tasks.table_extraction import TableExtractionTask
 from nv_ingest_client.primitives.tasks.chart_extraction import ChartExtractionTask
 from nv_ingest_client.primitives.tasks.infographic_extraction import InfographicExtractionTask
+from nv_ingest.api.dependency import get_ingest_service
 
 logger = logging.getLogger("uvicorn")
 tracer = trace.get_tracer(__name__)
 
 router = APIRouter()
 
 
-async def _get_ingest_service() -> IngestServiceMeta:
-    """
-    Gather the appropriate Ingestion Service to use for the nv-ingest endpoint.
-    """
-    logger.debug("Creating RedisIngestService singleton for dependency injection")
-    return RedisIngestService.get_instance()
-
-
-INGEST_SERVICE_T = Annotated[IngestServiceMeta, Depends(_get_ingest_service)]
+INGEST_SERVICE_T = Annotated[IngestServiceMeta, Depends(get_ingest_service)]
 STATE_RETRIEVED_DESTRUCTIVE = "RETRIEVED_DESTRUCTIVE"
 STATE_RETRIEVED_NON_DESTRUCTIVE = "RETRIEVED_NON_DESTRUCTIVE"
 STATE_RETRIEVED_CACHED = "RETRIEVED_CACHED"