"""Starlette application factory for ollama-web.""" from __future__ import annotations import logging from collections.abc import AsyncIterator from contextlib import asynccontextmanager from pathlib import Path from typing import Any from starlette.applications import Starlette from starlette.middleware import Middleware from starlette.middleware.cors import CORSMiddleware from starlette.responses import Response from starlette.routing import Mount, Route from starlette.staticfiles import StaticFiles from .auth import AuthMiddleware from .config import settings from .middleware import SecurityHeadersMiddleware from .routes import auth as auth_route from .routes import chat as chat_route from .routes import mcp as mcp_route from .routes import models as models_route from .routes import pages as pages_route from .routes import sessions as sessions_route from .routes import settings_route from .sessions import SessionStore class NoCacheStaticFiles(StaticFiles): """StaticFiles variant that disables browser caching for local development. This ensures language changes and JS updates are reflected immediately after a server restart, without requiring users to clear the browser cache. """ async def get_response(self, path: str, scope: Any) -> Response: response = await super().get_response(path, scope) response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0" return response _STATIC_DIR = Path(__file__).resolve().parent / "static" @asynccontextmanager async def _lifespan(app: Starlette) -> AsyncIterator[None]: """Keep the application log handler scoped to the ASGI lifespan.""" log_dir = Path(app.state.settings.data_dir).resolve() log_dir.mkdir(parents=True, exist_ok=True) handler = logging.FileHandler(str(log_dir / "server.log"), encoding="utf-8") handler.setFormatter( logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s") ) logger = logging.getLogger("ollama_web") previous_level = logger.level logger.setLevel(logging.INFO) logger.addHandler(handler) try: yield finally: logger.removeHandler(handler) handler.close() logger.setLevel(previous_level) def create_app() -> Starlette: """Build and return the configured Starlette application.""" routes = [ Route("/login", pages_route.login, name="login_page"), Route("/api/auth/login", auth_route.login, methods=["POST"], name="auth_login"), Route("/api/auth/logout", auth_route.logout, methods=["POST"], name="auth_logout"), Route("/api/auth/status", auth_route.status, methods=["GET"], name="auth_status"), Route("/", pages_route.index, name="index"), Route("/api/chat", chat_route.chat, methods=["POST"], name="chat"), Route("/api/models", models_route.get_models, methods=["GET"], name="models"), Route( "/api/models/{model:path}/capabilities", models_route.get_model_capabilities, methods=["GET"], name="model_capabilities", ), Route("/api/health", models_route.health, methods=["GET"], name="health"), Route( "/api/settings", settings_route.get_settings, methods=["GET"], name="settings", ), Route( "/api/settings", settings_route.put_settings, methods=["PUT"], name="settings_update", ), Route("/api/sessions", sessions_route.list_sessions, methods=["GET"], name="list_sessions"), Route( "/api/sessions", sessions_route.create_session, methods=["POST"], name="create_session", ), Route( "/api/sessions/{session_id}", sessions_route.get_session, methods=["GET"], name="get_session", ), Route( "/api/sessions/{session_id}", sessions_route.delete_session, methods=["DELETE"], name="delete_session", ), Route( "/api/sessions/{session_id}/files", sessions_route.upload_file, methods=["POST"], name="upload_file", ), Route( "/api/sessions/{session_id}/files/{file_id}", sessions_route.delete_file, methods=["DELETE"], name="delete_file", ), Route( "/api/sessions/{session_id}/files/{file_id}/content", sessions_route.get_file_content, methods=["GET", "HEAD"], name="get_file_content", ), Route( "/api/sessions/{session_id}/messages", sessions_route.clear_messages, methods=["DELETE"], name="clear_messages", ), Route("/api/mcp/servers", mcp_route.get_servers, methods=["GET"], name="mcp_servers"), Route( "/api/mcp/servers", mcp_route.put_servers, methods=["PUT"], name="mcp_servers_update", ), Mount("/static", app=NoCacheStaticFiles(directory=str(_STATIC_DIR)), name="static"), ] middleware = [Middleware(SecurityHeadersMiddleware), Middleware(AuthMiddleware)] if settings.allowed_origins: middleware.append( Middleware( CORSMiddleware, allow_origins=settings.allowed_origins, allow_methods=["GET", "POST", "PUT", "DELETE"], allow_headers=["Content-Type", "X-CSRF-Token"], allow_credentials=True, ) ) app = Starlette(routes=routes, middleware=middleware, lifespan=_lifespan) app.state.settings = settings app.state.session_store = SessionStore(settings.data_dir) app.state.tool_names = ["web_search", "scrape_url", "search_and_fetch", "pdf_to_text"] return app app = create_app()