Spaces:
Starting
Starting
File size: 3,234 Bytes
3326079 d01a7e3 3326079 d01a7e3 3326079 d01a7e3 3326079 d01a7e3 3326079 d01a7e3 3326079 d01a7e3 3326079 d01a7e3 3326079 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 |
"""FastAPI application module for the RAG chatbot.
This module provides the FastAPI application and routing for the
RAG chatbot API. The API supports:
- Query endpoint with SSE streaming
- Health check endpoint
- Error handling middleware
Components:
- create_app: Factory function for FastAPI application
- query router: Handles chat queries
- health router: Provides health checks
- SSE utilities: Server-Sent Events streaming support
Lazy Loading:
FastAPI and related dependencies are loaded on first access
using __getattr__. This ensures fast import times when the
API module is not immediately needed.
Example:
-------
>>> from rag_chatbot.api import create_app
>>> app = create_app()
>>> # Run with: uvicorn rag_chatbot.api:app
SSE Streaming Example:
-------
>>> from rag_chatbot.api import stream_sse_response, SourceInfo
>>> from rag_chatbot.llm.base import LLMRequest
>>>
>>> # Stream LLM response as SSE events
>>> async for event in stream_sse_response(registry, request, sources):
... yield event
"""
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .artifact_downloader import ArtifactDownloader, ArtifactDownloadError
from .main import create_app
from .sse import (
DoneEvent,
ErrorEvent,
SourceInfo,
TokenEvent,
format_sse_event,
stream_sse_response,
)
# =============================================================================
# Module Exports
# =============================================================================
__all__: list[str] = [
"create_app",
# SSE utilities
"TokenEvent",
"DoneEvent",
"ErrorEvent",
"SourceInfo",
"stream_sse_response",
"format_sse_event",
# Artifact downloader
"ArtifactDownloader",
"ArtifactDownloadError",
]
def __getattr__(name: str) -> object:
"""Lazy load module exports on first access.
This function is called when an attribute is not found in the module's
namespace. It enables lazy loading of FastAPI and related dependencies.
Args:
----
name: The name of the attribute being accessed.
Returns:
-------
The requested attribute if it exists in __all__.
Raises:
------
AttributeError: If the attribute is not a valid export.
"""
# Main app factory
if name == "create_app":
from .main import create_app
return create_app
# SSE utilities - use a mapping to reduce return statements
sse_exports = {
"TokenEvent",
"DoneEvent",
"ErrorEvent",
"SourceInfo",
"stream_sse_response",
"format_sse_event",
}
if name in sse_exports:
from . import sse
return getattr(sse, name)
# Artifact downloader exports
artifact_exports = {
"ArtifactDownloader",
"ArtifactDownloadError",
}
if name in artifact_exports:
from . import artifact_downloader
return getattr(artifact_downloader, name)
msg = f"module {__name__!r} has no attribute {name!r}" # pragma: no cover
raise AttributeError(msg) # pragma: no cover
|