refactor into functional core/imperative shell

src/fastcache_api/core.py

@@ -0,0 +1,28 @@
+"""Pure decision logic: no IO, no DB, no subprocess access."""
+
+from collections.abc import Iterable
+
+from .models import CacheConfig, CacheState
+
+NON_FINAL_STATES: list[str] = [s.value for s in CacheState if not s.is_final()]
+
+
+def decide_final_state(exit_code: int | None) -> CacheState:
+    return CacheState.completed if exit_code == 0 else CacheState.failed
+
+
+def ports_in_use(configs: Iterable[CacheConfig]) -> set[int]:
+    used: set[int] = set()
+    for config in configs:
+        for uri in (config.pull_uri, config.push_uri):
+            if uri.port is not None:
+                used.add(uri.port)
+    return used
+
+
+def allocate_port_pair(in_use: set[int], start: int, end: int) -> tuple[int, int]:
+    for pull in range(start, end + 1, 2):
+        push = pull + 1
+        if push <= end and pull not in in_use and push not in in_use:
+            return pull, push
+    raise RuntimeError(f"no free cache port pair in range [{start}, {end}]")

src/fastcache_api/exceptions.py

@@ -0,0 +1,54 @@
+from fastapi import FastAPI, Request, status
+from fastapi.responses import JSONResponse
+
+
+class CacheNotFound(Exception):
+    pass
+
+
+class CacheKeyConflict(Exception):
+    pass
+
+
+class CachePortsExhausted(Exception):
+    pass
+
+
+class CacheStartFailed(Exception):
+    pass
+
+
+def register_exception_handlers(app: FastAPI) -> None:
+    @app.exception_handler(CacheNotFound)
+    async def _handle_not_found(_request: Request, exc: CacheNotFound) -> JSONResponse:
+        return JSONResponse(
+            status_code=status.HTTP_404_NOT_FOUND,
+            content={"detail": str(exc) or "Cache not found"},
+        )
+
+    @app.exception_handler(CacheKeyConflict)
+    async def _handle_key_conflict(
+        _request: Request, exc: CacheKeyConflict
+    ) -> JSONResponse:
+        return JSONResponse(
+            status_code=status.HTTP_409_CONFLICT,
+            content={"detail": str(exc) or "Cache key conflict"},
+        )
+
+    @app.exception_handler(CachePortsExhausted)
+    async def _handle_ports_exhausted(
+        _request: Request, exc: CachePortsExhausted
+    ) -> JSONResponse:
+        return JSONResponse(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            content={"detail": str(exc) or "No free cache ports"},
+        )
+
+    @app.exception_handler(CacheStartFailed)
+    async def _handle_start_failed(
+        _request: Request, exc: CacheStartFailed
+    ) -> JSONResponse:
+        return JSONResponse(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            content={"detail": str(exc) or "Failed to start cache process"},
+        )

src/fastcache_api/main.py

@@ -10,6 +10,7 @@
 from fastapi.routing import APIRoute
 
 from .config import settings
+from .exceptions import register_exception_handlers
 from .lifecycle import exit_watchers
 from .reconcile import monitor_caches, reconcile_caches
 from .routes import api_router
@@ -42,6 +43,8 @@ async def lifespan(application: FastAPI) -> AsyncGenerator[None]:
     lifespan=lifespan,
 )
 
+register_exception_handlers(app)
+
 app.include_router(api_router, prefix=settings.API_V1_STR)
 
 

src/fastcache_api/process.py

@@ -2,7 +2,6 @@
 import logging
 import socket
 import subprocess
-from collections.abc import Iterable
 from pathlib import Path
 from uuid import UUID
 
@@ -21,23 +20,6 @@ def canonical_hostname() -> str:
     return (socket.getfqdn() or socket.gethostname()).lower()
 
 
-def ports_in_use(configs: Iterable[CacheConfig]) -> set[int]:
-    used: set[int] = set()
-    for config in configs:
-        for uri in (config.pull_uri, config.push_uri):
-            if uri.port is not None:
-                used.add(uri.port)
-    return used
-
-
-def allocate_port_pair(in_use: set[int], start: int, end: int) -> tuple[int, int]:
-    for pull in range(start, end + 1, 2):
-        push = pull + 1
-        if push <= end and pull not in in_use and push not in in_use:
-            return pull, push
-    raise RuntimeError(f"no free cache port pair in range [{start}, {end}]")
-
-
 # Live anyio Process handles for children we spawned
 _processes: dict[int, anyio.abc.Process] = {}
 

src/fastcache_api/reconcile.py

@@ -3,19 +3,15 @@
 from uuid import UUID
 
 import anyio
-from sqlalchemy import select
 
+from . import repo
 from .config import settings
 from .db import SessionLocal
 from .models import CacheState
 from .process import exit_code, is_alive, wait_exit
-from .tables import Cache
 
 logger = logging.getLogger(__name__)
 
-# States whose process should still be running, so must be checked vs reality.
-_NON_FINAL = [s.value for s in CacheState if not s.is_final()]
-
 
 async def sweep_dead_caches() -> int:
     """Reconcile non-final caches whose process is gone; return how many changed.
@@ -25,17 +21,14 @@ async def sweep_dead_caches() -> int:
     api restart) -> failed.
     """
     async with SessionLocal() as session:
-        result = await session.execute(select(Cache).where(Cache.state.in_(_NON_FINAL)))
-        caches = result.scalars().all()
+        caches = await repo.list_non_final(session)
 
         stale = 0
         for cache in caches:
             if is_alive(cache.pid, cache.create_time):
                 continue
             ec = exit_code(cache.pid)
-            cache.state = CacheState.completed if ec == 0 else CacheState.failed
-            cache.exit_code = ec
-            cache.key = None  # Free the key!
+            repo.finalize_cache(cache, ec)
             logger.warning(
                 "Cache %s (pid=%d) is no longer running (exit_code=%s); marking %s",
                 cache.id,
@@ -64,12 +57,10 @@ async def watch_and_record(cache_id: UUID, pid: int) -> None:
     # and this won't be able to commit properly
     with anyio.CancelScope(shield=True):
         async with SessionLocal() as session:
-            cache = await session.get(Cache, cache_id)
+            cache = await repo.get_cache(session, cache_id)
             if cache is None or CacheState(cache.state).is_final():
                 return
-            cache.state = CacheState.completed if exit_code == 0 else CacheState.failed
-            cache.exit_code = exit_code
-            cache.key = None
+            repo.finalize_cache(cache, exit_code)
             await session.commit()
     logger.info(
         "Cache %s (pid=%d) exited (code=%s); marked %s",

src/fastcache_api/repo.py

@@ -0,0 +1,51 @@
+from collections.abc import Sequence
+from uuid import UUID
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from .core import NON_FINAL_STATES, decide_final_state
+from .models import CacheConfig
+from .tables import Cache
+
+
+async def get_cache(session: AsyncSession, cache_id: UUID) -> Cache | None:
+    return await session.get(Cache, cache_id)
+
+
+async def list_caches(session: AsyncSession) -> Sequence[Cache]:
+    result = await session.execute(select(Cache))
+    return result.scalars().all()
+
+
+async def list_non_final(session: AsyncSession) -> Sequence[Cache]:
+    result = await session.execute(
+        select(Cache).where(Cache.state.in_(NON_FINAL_STATES))
+    )
+    return result.scalars().all()
+
+
+async def find_active_by_key(session: AsyncSession, key: str) -> Cache | None:
+    result = await session.execute(
+        select(Cache).where(Cache.key == key, Cache.state.in_(NON_FINAL_STATES))
+    )
+    return result.scalar_one_or_none()
+
+
+async def list_active_configs(session: AsyncSession) -> list[CacheConfig]:
+    caches = await list_non_final(session)
+    return [CacheConfig.model_validate(cache.config) for cache in caches]
+
+
+def insert_cache(session: AsyncSession, cache: Cache) -> None:
+    session.add(cache)
+
+
+def finalize_cache(cache: Cache, exit_code: int | None) -> None:
+    """Set `cache` to its terminal state for `exit_code` and free its key.
+
+    Does not commit; caller owns the transaction boundary.
+    """
+    cache.state = decide_final_state(exit_code)
+    cache.exit_code = exit_code
+    cache.key = None