diff --git a/app/services/__init__.py b/app/services/__init__.py index e69de29..a0dac8d 100644 --- a/app/services/__init__.py +++ b/app/services/__init__.py @@ -0,0 +1 @@ +"""Application service implementations.""" diff --git a/app/services/auth.py b/app/services/auth.py index 0a4f973..6499879 100644 --- a/app/services/auth.py +++ b/app/services/auth.py @@ -1,3 +1,5 @@ +"""Bearer-token authentication and Redis-backed token storage.""" + import time import jwt @@ -11,24 +13,27 @@ from app.models.user import User logger = get_logger() -async def get_from_redis(request: Request, key: str): +async def get_from_redis(request: Request, key: str) -> str | None: + """Return the Redis value stored under ``key``.""" return await request.app.redis.get(key) -async def set_to_redis(request: Request, key: str, value: str, ex: int): +async def set_to_redis(request: Request, key: str, value: str, ex: int) -> bool: + """Store ``value`` in Redis for ``ex`` seconds.""" return await request.app.redis.set(key, value, ex=ex) async def verify_jwt(request: Request, token: str) -> bool: + """Check whether Redis recognizes the bearer token.""" payload = await get_from_redis(request, token) return bool(payload) class AuthBearer(HTTPBearer): - def __init__(self, auto_error: bool = True) -> None: - super().__init__(auto_error=auto_error) + """Validate bearer credentials against the active Redis token store.""" - async def __call__(self, request: Request): + async def __call__(self, request: Request) -> str: + """Return a valid token or raise an authorization error.""" credentials: HTTPAuthorizationCredentials = await super().__call__(request) if not credentials: raise HTTPException(status_code=403, detail="Invalid authorization code.") @@ -44,8 +49,8 @@ class AuthBearer(HTTPBearer): return credentials.credentials -async def create_access_token(user: User, request: Request): - # sourcery skip: avoid-builtin-shadow +async def create_access_token(user: User, request: Request) -> str | None: + """Sign a user token and store it in Redis until its configured expiry.""" payload = { "email": user.email, "expiry": time.time() + global_settings.jwt_expire, @@ -55,8 +60,8 @@ async def create_access_token(user: User, request: Request): payload, str(user.password), algorithm=global_settings.jwt_algorithm ) - _bool = await set_to_redis( + stored = await set_to_redis( request, token, str(payload), ex=global_settings.jwt_expire ) - if _bool: + if stored: return token diff --git a/app/services/chat_agent.py b/app/services/chat_agent.py index 7ef5725..6a237b6 100644 --- a/app/services/chat_agent.py +++ b/app/services/chat_agent.py @@ -16,7 +16,7 @@ from typing import Protocol, runtime_checkable from urllib.parse import urlparse import attrs -import httpx +import httpx2 import orjson from app.config import ChatConfig @@ -49,9 +49,11 @@ class LocalEchoAgent: """ def __init__(self, stream_delay_seconds: float = 0.02) -> None: + """Initialize the delay between streamed words.""" self.stream_delay_seconds = stream_delay_seconds def _compose_reply(self, messages: list[ChatMessage]) -> str: + """Compose a canned response to the latest user message.""" last_user = next( (m.content for m in reversed(messages) if m.role == "user"), "" ) @@ -63,26 +65,20 @@ class LocalEchoAgent: return f"You said: {last_user!r}. This is a local echo response (stub agent)." async def stream_reply(self, messages: list[ChatMessage]) -> AsyncIterator[str]: + """Yield the canned reply one word at a time.""" reply = self._compose_reply(messages) for word in reply.split(" "): await asyncio.sleep(self.stream_delay_seconds + random.uniform(0, 0.01)) yield word + " " async def aclose(self) -> None: - return None + """Close resources; the local agent holds none.""" -def _validate_base_url(_instance: OllamaChatAgent, _attribute: attrs.Attribute, value: str) -> None: - """Validate that base_url is a valid HTTP(S) URL. - - Args: - instance: The OllamaChatAgent instance being initialized. - attribute: The attrs attribute descriptor for base_url. - value: The URL string to validate. - - Raises: - ValueError: If base_url is not a valid HTTP(S) URL. - """ +def _validate_base_url( + _instance: OllamaChatAgent, _attribute: attrs.Attribute, value: str +) -> None: + """Require an absolute HTTP(S) endpoint URL.""" parsed = urlparse(value) if parsed.scheme not in ("http", "https"): msg = f"base_url must be HTTP(S), got scheme '{parsed.scheme}' from '{value}'" @@ -92,37 +88,15 @@ def _validate_base_url(_instance: OllamaChatAgent, _attribute: attrs.Attribute, raise ValueError(msg) -def _validate_timeout(_instance: OllamaChatAgent, _attribute: attrs.Attribute, value: float) -> None: - """Validate that timeout is positive. - - Args: - instance: The OllamaChatAgent instance being initialized. - attribute: The attrs attribute descriptor for timeout. - value: The timeout value in seconds. - - Raises: - ValueError: If timeout is not positive. - """ +def _validate_timeout( + _instance: OllamaChatAgent, _attribute: attrs.Attribute, value: float +) -> None: + """Require a positive request timeout.""" if value <= 0: msg = f"timeout must be positive, got {value}" raise ValueError(msg) -def _create_httpx_client(instance: OllamaChatAgent) -> httpx.AsyncClient: - """Factory function to create the httpx.AsyncClient with validated config. - - This function is called from __attrs_post_init__ after all field validators - have run, ensuring the client is created with validated configuration. - - Args: - instance: The OllamaChatAgent instance being initialized. - - Returns: - An initialized httpx.AsyncClient configured with base_url and timeout. - """ - return httpx.AsyncClient(base_url=instance.base_url, timeout=instance.timeout) - - @attrs.define(slots=True, eq=False, hash=False) class OllamaChatAgent: """Streams chat completions from an OpenAI-compatible endpoint. @@ -132,9 +106,6 @@ class OllamaChatAgent: a ready-to-swap-in replacement for :class:`LocalEchoAgent` once a real model should be used. - Attrs Configuration: - - slots=True: Memory-efficient attribute storage (~40-50% reduction) - - eq=False, hash=False: Instances are not comparable (contain async resources) """ model: str = attrs.field( @@ -160,20 +131,15 @@ class OllamaChatAgent: "constraints": "Must be positive", }, ) - _client: httpx.AsyncClient = attrs.field( + _client: httpx2.AsyncClient = attrs.field( init=False, repr=False, metadata={"description": "Internal HTTP client for API communication"}, ) def __attrs_post_init__(self) -> None: - """Initialize the HTTP client after field validation. - - This hook is called by attrs after __init__ completes and all field - validators have run. It's used to initialize the internal _client - field which depends on validated configuration. - """ - self._client = _create_httpx_client(self) + """Create the HTTPX2 client after attrs validates the fields.""" + self._client = httpx2.AsyncClient(base_url=self.base_url, timeout=self.timeout) async def stream_reply(self, messages: list[ChatMessage]) -> AsyncIterator[str]: """Stream chat completion responses from the configured model. @@ -195,13 +161,11 @@ class OllamaChatAgent: async for line in response.aiter_lines(): if not line.startswith("data: ") or line == "data: [DONE]": continue - try: - data = orjson.loads(line[6:]) - content = ( - data.get("choices", [{}])[0].get("delta", {}).get("content", "") - ) - except Exception: - content = "" + data = orjson.loads(line.removeprefix("data: ")) + choices = data.get("choices", []) + if not choices: + continue + content = choices[0].get("delta", {}).get("content", "") if content: yield content diff --git a/app/services/chat_session.py b/app/services/chat_session.py index 92d87f4..8b8b3bd 100644 --- a/app/services/chat_session.py +++ b/app/services/chat_session.py @@ -17,13 +17,17 @@ from app.schemas.chat import ChatMessage @attrs.define(slots=True) class ChatSession: + """Conversation history associated with one websocket connection.""" + id: UUID = attrs.field(factory=uuid4) messages: list[ChatMessage] = attrs.field(factory=list) def add(self, message: ChatMessage) -> None: + """Append a message to this conversation.""" self.messages.append(message) def history(self) -> list[ChatMessage]: + """Return a copy of the conversation history.""" return list(self.messages) @@ -35,23 +39,28 @@ class ChatSessionManager: """ def __init__(self) -> None: + """Create an empty, lock-protected session registry.""" self._sessions: dict[UUID, ChatSession] = {} self._lock = asyncio.Lock() async def create(self) -> ChatSession: + """Create and register a new conversation.""" session = ChatSession() async with self._lock: self._sessions[session.id] = session return session async def get(self, session_id: UUID) -> ChatSession | None: + """Return the matching session, if it is active.""" async with self._lock: return self._sessions.get(session_id) async def remove(self, session_id: UUID) -> None: + """Remove a session from the active registry.""" async with self._lock: self._sessions.pop(session_id, None) async def count(self) -> int: + """Return the number of active sessions.""" async with self._lock: return len(self._sessions) diff --git a/app/services/llm.py b/app/services/llm.py index ceaa455..cc1f4a3 100644 --- a/app/services/llm.py +++ b/app/services/llm.py @@ -1,25 +1,28 @@ -from collections.abc import AsyncGenerator +"""Streaming client for OpenAI-compatible chat completion endpoints.""" -import httpx +from collections.abc import AsyncIterator + +import httpx2 import orjson class StreamLLMService: + """Send prompts to an LLM endpoint and yield newline-delimited JSON events.""" + def __init__(self, base_url: str = "http://localhost:11434/v1") -> None: + """Configure the endpoint and default model.""" self.base_url = base_url self.model = "llama3.2" - async def stream_chat(self, prompt: str) -> AsyncGenerator[bytes]: - """Stream chat completion responses from LLM.""" - # Send the user a message first + async def stream_chat(self, prompt: str) -> AsyncIterator[bytes]: + """Yield the prompt followed by streamed model response chunks.""" user_msg = { "role": "user", "content": prompt, } yield orjson.dumps(user_msg) + b"\n" - # Open client as context manager and stream responses - async with httpx.AsyncClient(base_url=self.base_url) as client: + async with httpx2.AsyncClient(base_url=self.base_url) as client: async with client.stream( "POST", "/chat/completions", @@ -32,21 +35,16 @@ class StreamLLMService: ) as response: async for line in response.aiter_lines(): if line.startswith("data: ") and line != "data: [DONE]": - try: - json_line = line[6:] # Remove "data: " prefix - data = orjson.loads(json_line) - content = ( - data.get("choices", [{}])[0] - .get("delta", {}) - .get("content", "") - ) - if content: - model_msg = {"role": "model", "content": content} - yield orjson.dumps(model_msg) + b"\n" - except Exception: - pass + data = orjson.loads(line.removeprefix("data: ")) + choices = data.get("choices", []) + if not choices: + continue + content = choices[0].get("delta", {}).get("content", "") + if content: + model_msg = {"role": "model", "content": content} + yield orjson.dumps(model_msg) + b"\n" -# FastAPI dependency def get_llm_service(base_url: str | None = None) -> StreamLLMService: + """Build the service using the supplied URL or its local default.""" return StreamLLMService(base_url=base_url or "http://localhost:11434/v1") diff --git a/app/services/scheduler.py b/app/services/scheduler.py index 22182ce..be2ff5c 100644 --- a/app/services/scheduler.py +++ b/app/services/scheduler.py @@ -1,3 +1,5 @@ +"""Background database health-check scheduling.""" + from collections.abc import AsyncIterator from contextlib import asynccontextmanager from datetime import datetime @@ -15,7 +17,8 @@ from app.database import AsyncSessionFactory, engine logger = get_logger() -async def tick(): +async def tick() -> bool: + """Run the scheduler's lightweight database heartbeat.""" async with AsyncSessionFactory() as session: stmt = text("select 1;") await logger.ainfo(f">>>> Be or not to be...{datetime.now()}") @@ -26,6 +29,7 @@ async def tick(): @asynccontextmanager async def scheduler_lifespan() -> AsyncIterator[None]: + """Start the scheduler and stop it when the app lifespan ends.""" scheduler = AsyncScheduler( SQLAlchemyDataStore(engine, schema="scheduler"), RedisEventBroker(client_or_url=global_settings.redis_url.unicode_string()), diff --git a/app/services/smtp.py b/app/services/smtp.py index 3d09932..16734f7 100644 --- a/app/services/smtp.py +++ b/app/services/smtp.py @@ -1,3 +1,5 @@ +"""Email delivery through a configured SMTP server.""" + import smtplib from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText diff --git a/pyproject.toml b/pyproject.toml index c76e5b3..966be12 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,7 +12,7 @@ dependencies = [ "uvicorn[standard]==0.54.0", "asyncpg==0.32.0", "alembic==1.20.1", - "httpx==0.28.1", + "httpx2==2.16.0", "uvloop==0.23.0", "httptools==0.9.0", "rich==15.0.0", @@ -47,6 +47,7 @@ test = [ "inline-snapshot==0.36.1", "dirty-equals==0.11", "polyfactory==3.5.0", + "httpx==0.28.1", "httpx2[ws]==2.16.0", ] diff --git a/tests/test_ollama_chat_agent_attrs.py b/tests/test_ollama_chat_agent_attrs.py index 19250f2..3866677 100644 --- a/tests/test_ollama_chat_agent_attrs.py +++ b/tests/test_ollama_chat_agent_attrs.py @@ -1,5 +1,6 @@ """Test suite for refactored OllamaChatAgent with attrs features.""" -import httpx + +import httpx2 import pytest from app.config import ChatConfig @@ -54,14 +55,14 @@ class TestOllamaChatAgentAttrsFeatures: assert isinstance(agent.timeout, float) assert agent.timeout == 30.0 - def test_factory_initialization_of_client(self): - """Test that _client is created via factory function.""" + def test_client_initialization_after_field_validation(self): + """Test that the async client uses the validated configuration.""" agent = OllamaChatAgent(model="llama3.2", base_url="http://localhost:11434/v1") assert hasattr(agent, "_client") - assert isinstance(agent._client, httpx.AsyncClient) - # httpx normalizes URLs by adding a trailing slash + assert isinstance(agent._client, httpx2.AsyncClient) + # httpx2 normalizes URLs by adding a trailing slash assert str(agent._client.base_url) == "http://localhost:11434/v1/" - assert agent._client.timeout == httpx.Timeout(60.0) + assert agent._client.timeout == httpx2.Timeout(60.0) def test_slots_enabled_no_dict(self): """Test that slots=True prevents __dict__ attribute.""" @@ -72,12 +73,8 @@ class TestOllamaChatAgentAttrsFeatures: def test_equality_disabled(self): """Test that eq=False means instances are not equal even with same values.""" - agent1 = OllamaChatAgent( - model="llama3.2", base_url="http://localhost:11434/v1" - ) - agent2 = OllamaChatAgent( - model="llama3.2", base_url="http://localhost:11434/v1" - ) + agent1 = OllamaChatAgent(model="llama3.2", base_url="http://localhost:11434/v1") + agent2 = OllamaChatAgent(model="llama3.2", base_url="http://localhost:11434/v1") # With eq=False, only identity comparison works assert agent1 != agent2 assert agent1 == agent1 @@ -112,9 +109,7 @@ class TestOllamaChatAgentAttrsFeatures: def test_https_urls_accepted(self): """Test that HTTPS URLs are properly accepted.""" - agent = OllamaChatAgent( - model="llama3.2", base_url="https://api.example.com/v1" - ) + agent = OllamaChatAgent(model="llama3.2", base_url="https://api.example.com/v1") assert agent.base_url == "https://api.example.com/v1" def test_metadata_annotations_present(self): @@ -133,4 +128,3 @@ class TestOllamaChatAgentAttrsFeatures: if __name__ == "__main__": pytest.main([__file__, "-v"]) - diff --git a/uv.lock b/uv.lock index 41ffe86..ce7e279 100644 --- a/uv.lock +++ b/uv.lock @@ -330,7 +330,7 @@ dependencies = [ { name = "fastexcel" }, { name = "granian" }, { name = "httptools" }, - { name = "httpx" }, + { name = "httpx2" }, { name = "jinja2" }, { name = "polars" }, { name = "pydantic", extra = ["email"] }, @@ -359,6 +359,7 @@ dev = [ ] test = [ { name = "dirty-equals" }, + { name = "httpx" }, { name = "httpx2", extra = ["ws"] }, { name = "inline-snapshot" }, { name = "polyfactory" }, @@ -376,7 +377,7 @@ requires-dist = [ { name = "fastexcel", specifier = "==0.21.0" }, { name = "granian", specifier = "==2.8.5" }, { name = "httptools", specifier = "==0.9.0" }, - { name = "httpx", specifier = "==0.28.1" }, + { name = "httpx2", specifier = "==2.16.0" }, { name = "jinja2", specifier = "==3.1.6" }, { name = "polars", specifier = "==2.0.0" }, { name = "pydantic", extras = ["email"], specifier = "==2.14.1" }, @@ -405,6 +406,7 @@ dev = [ ] test = [ { name = "dirty-equals", specifier = "==0.11" }, + { name = "httpx", specifier = "==0.28.1" }, { name = "httpx2", extras = ["ws"], specifier = "==2.16.0" }, { name = "inline-snapshot", specifier = "==0.36.1" }, { name = "polyfactory", specifier = "==3.5.0" },