feat(perf): implement HTTP connection pooling with httpx.AsyncClient singleton and Keep-Alive

- get_http_client: singleton AsyncClient with limits(max_keepalive=10, max_connections=20, keepalive_expiry=60s)
- close_http_client: clean socket pool teardown in FastAPI lifespan shutdown
- refactor dictionary fetchers to reuse pooled client and avoid 5x TLS handshakes per word lookup
- add test_http_connection_pooling (22/22 pytest passed)
This commit is contained in:
Antigravity committed 2026-10-06 11:34:31 +03:00
1 parent c4a6edee37
commit 438dfdb055
4 files changed
+211 -132

No files matched your search

+19 -5
View File
@@ -175,13 +175,27 @@
--- ---
### 8. 🔌 Пул соединений HTTP (`httpx.AsyncClient` с Keep-Alive) ### 7.1. 🖼️ Извлечение и отображение схем, топологий и диаграмм из PDF / EPUB (Planned)
- [ ] В [`app/dictionary.py`](file:///home/ai-projects/services/study.qpid.pro/app/dictionary.py) создать синглтон-клиент `get_http_client()` с пулом соединений: - [ ] **Извлечение бинарных изображений на диск**:
- При парсинге PDF/EPUB извлекать графические объекты (`page.images` / архивы изображений) на файловую систему в изолированную директорию книги: `data/books/{book_id}/images/{img_name}`.
- Избегать раздувания базы данных SQLite: база хранит только текст и метаданные, бинарная графика хранится на диске.
- [ ] **Кэширующий медиа-эндпоинт**:
- Создать эндпоинт `GET /api/books/{book_id}/images/{image_name}` с заголовками кэширования (`Cache-Control: public, max-age=31536000, immutable`).
- [ ] **Встраивание в текст читалки**:
- Привязывать иллюстрации к соответствующей странице/главе через Markdown-теги: `![Схема топологии](/api/books/{book_id}/images/{img_name})`.
- Интерактивный просмотр и зум схем по клику в читалке.
---
### 8. 🔌 Пул соединений HTTP (`httpx.AsyncClient` с Keep-Alive) (Completed)
- [x] В [`app/dictionary.py`](file:///home/ai-projects/services/study.qpid.pro/app/dictionary.py) создать синглтон-клиент `get_http_client()` с пулом соединений:
```python ```python
limits = httpx.Limits(max_keepalive_connections=10, max_connections=20) limits = httpx.Limits(max_keepalive_connections=10, max_connections=20, keepalive_expiry=60.0)
``` ```
- [ ] Заменить вызовы `async with httpx.AsyncClient()` в парсерах на переиспользуемый клиент. - [x] Заменить вызовы `async with httpx.AsyncClient()` в парсерах на переиспользуемый клиент.
- [ ] Устранить 5 повторных TCP/TLS хэндшейков на каждое слово, сократив пинг внешних запросов на 300–600 мс. - [x] Устранить 5 повторных TCP/TLS хэндшейков на каждое слово, сократив задержку внешних запросов на 300–600 мс.
- [x] Добавить вызов `close_http_client()` в `lifespan` FastAPI при остановке приложения для чистого закрытия открытых сокетов.
- [x] Добавить юнит-тесты `test_http_connection_pooling` в `tests/test_api.py` (22/22 passed).
--- ---
+159 -126
View File
@@ -8,7 +8,39 @@ from app.database import get_db
USER_AGENT = "NotesApp/1.0 (https://notes.qpid.pro; admin@qpid.pro)" USER_AGENT = "NotesApp/1.0 (https://notes.qpid.pro; admin@qpid.pro)"
async def fetch_wiktionary_wikitext(word: str) -> Dict[str, Any]: _http_client: Optional[httpx.AsyncClient] = None
def get_http_client() -> httpx.AsyncClient:
"""
Returns a shared singleton httpx.AsyncClient instance with connection pooling and Keep-Alive.
Reuses persistent TCP/TLS connections to dictionary providers (Wiktionary, Datamuse, MyMemory)
eliminating 300-600ms of repeated connection handshakes per word lookup.
"""
global _http_client
if _http_client is None or _http_client.is_closed:
limits = httpx.Limits(max_keepalive_connections=10, max_connections=20, keepalive_expiry=60.0)
timeout = httpx.Timeout(connect=3.0, read=4.0, write=3.0, pool=3.0)
_http_client = httpx.AsyncClient(
limits=limits,
timeout=timeout,
headers={"User-Agent": USER_AGENT},
follow_redirects=True
)
return _http_client
async def close_http_client():
"""
Closes the shared singleton HTTP client on application shutdown.
"""
global _http_client
if _http_client is not None and not _http_client.is_closed:
await _http_client.aclose()
_http_client = None
async def fetch_wiktionary_wikitext(word: str, client: Optional[httpx.AsyncClient] = None) -> Dict[str, Any]:
url = "https://en.wiktionary.org/w/api.php" url = "https://en.wiktionary.org/w/api.php"
params = { params = {
"action": "parse", "action": "parse",
@@ -17,111 +49,109 @@ async def fetch_wiktionary_wikitext(word: str) -> Dict[str, Any]:
"format": "json" "format": "json"
} }
try: try:
async with httpx.AsyncClient(timeout=4.0) as client: http = client or get_http_client()
resp = await client.get(url, params=params, headers={"User-Agent": USER_AGENT}) resp = await http.get(url, params=params, timeout=4.0)
if resp.status_code == 200: if resp.status_code == 200:
data = resp.json() data = resp.json()
wikitext = data.get("parse", {}).get("wikitext", {}).get("*", "") wikitext = data.get("parse", {}).get("wikitext", {}).get("*", "")
# Extract IPA
ipa_raw = re.findall(r"\{\{IPA\|en\|([^}]+)\}\}", wikitext)
ipa_cleaned = []
for entry in ipa_raw:
parts = entry.split("|")
for p in parts:
if p.startswith("/"):
ipa_cleaned.append(p)
# Extract Audio file names
audio_matches = re.findall(r"\{\{audio\|en\|([^|}]+)", wikitext)
# Extract Russian translations
ru_raw = re.findall(r"\{\{t[+-]?\|ru\|([^|}]+)", wikitext)
ru_cleaned = []
for r in ru_raw:
# Clean wiki markup like [[word]] or stress marks
clean = r.replace("[[", "").replace("]]", "").strip()
if clean and clean not in ru_cleaned:
ru_cleaned.append(clean)
return { # Extract IPA
"ipa": ipa_cleaned, ipa_raw = re.findall(r"\{\{IPA\|en\|([^}]+)\}\}", wikitext)
"audio": [], ipa_cleaned = []
"ru_translations": ru_cleaned for entry in ipa_raw:
} parts = entry.split("|")
for p in parts:
if p.startswith("/"):
ipa_cleaned.append(p)
# Extract Russian translations
ru_raw = re.findall(r"\{\{t[+-]?\|ru\|([^|}]+)", wikitext)
ru_cleaned = []
for r in ru_raw:
clean = r.replace("[[", "").replace("]]", "").strip()
if clean and clean not in ru_cleaned:
ru_cleaned.append(clean)
return {
"ipa": ipa_cleaned,
"audio": [],
"ru_translations": ru_cleaned
}
except Exception: except Exception:
pass pass
return {"ipa": [], "audio": [], "ru_translations": []} return {"ipa": [], "audio": [], "ru_translations": []}
async def fetch_wiktionary_definitions(word: str) -> Dict[str, Any]:
async def fetch_wiktionary_definitions(word: str, client: Optional[httpx.AsyncClient] = None) -> Dict[str, Any]:
url = f"https://en.wiktionary.org/api/rest_v1/page/definition/{urllib.parse.quote(word)}" url = f"https://en.wiktionary.org/api/rest_v1/page/definition/{urllib.parse.quote(word)}"
try: try:
async with httpx.AsyncClient(timeout=4.0) as client: http = client or get_http_client()
resp = await client.get(url, headers={"User-Agent": USER_AGENT}) resp = await http.get(url, timeout=4.0)
if resp.status_code == 200: if resp.status_code == 200:
data = resp.json() data = resp.json()
en_entries = data.get("en", []) en_entries = data.get("en", [])
parts_of_speech = [] parts_of_speech = []
definitions = [] definitions = []
examples = [] examples = []
for entry in en_entries: for entry in en_entries:
pos = entry.get("partOfSpeech") pos = entry.get("partOfSpeech")
if pos and pos not in parts_of_speech: if pos and pos not in parts_of_speech:
parts_of_speech.append(pos) parts_of_speech.append(pos)
for d in entry.get("definitions", []): for d in entry.get("definitions", []):
raw_def = d.get("definition", "") raw_def = d.get("definition", "")
# strip html tags clean_def = re.sub(r"<[^>]+>", "", raw_def).strip()
clean_def = re.sub(r"<[^>]+>", "", raw_def).strip() if clean_def and clean_def not in definitions:
if clean_def and clean_def not in definitions: definitions.append(clean_def)
definitions.append(clean_def) for ex in d.get("examples", []):
for ex in d.get("examples", []): clean_ex = re.sub(r"<[^>]+>", "", ex).strip()
clean_ex = re.sub(r"<[^>]+>", "", ex).strip() if clean_ex and clean_ex not in examples:
if clean_ex and clean_ex not in examples: examples.append(clean_ex)
examples.append(clean_ex) return {
return { "parts_of_speech": parts_of_speech,
"parts_of_speech": parts_of_speech, "definitions": definitions[:4],
"definitions": definitions[:4], "examples": examples[:3]
"examples": examples[:3] }
}
except Exception: except Exception:
pass pass
return {"parts_of_speech": [], "definitions": [], "examples": []} return {"parts_of_speech": [], "definitions": [], "examples": []}
async def fetch_datamuse(word: str) -> Dict[str, Any]:
async def fetch_datamuse(word: str, client: Optional[httpx.AsyncClient] = None) -> Dict[str, Any]:
url = "https://api.datamuse.com/words" url = "https://api.datamuse.com/words"
params = {"sp": word, "md": "drp", "ipa": "1"} params = {"sp": word, "md": "drp", "ipa": "1"}
try: try:
async with httpx.AsyncClient(timeout=3.0) as client: http = client or get_http_client()
resp = await client.get(url, params=params) resp = await http.get(url, params=params, timeout=3.0)
if resp.status_code == 200: if resp.status_code == 200:
items = resp.json() items = resp.json()
if items and isinstance(items, list): if items and isinstance(items, list):
first = items[0] first = items[0]
tags = first.get("tags", []) tags = first.get("tags", [])
ipa = None ipa = None
pos = None pos = None
for t in tags: for t in tags:
if t.startswith("ipa_pron:"): if t.startswith("ipa_pron:"):
ipa = "/" + t.split(":", 1)[1] + "/" ipa = "/" + t.split(":", 1)[1] + "/"
elif t in ("n", "v", "adj", "adv"): elif t in ("n", "v", "adj", "adv"):
pos_map = {"n": "Noun", "v": "Verb", "adj": "Adjective", "adv": "Adverb"} pos_map = {"n": "Noun", "v": "Verb", "adj": "Adjective", "adv": "Adverb"}
pos = pos_map.get(t) pos = pos_map.get(t)
defs = []
for d in first.get("defs", []):
parts = d.split("\t", 1)
defs.append(parts[1].strip() if len(parts) > 1 else parts[0].strip())
return { defs = []
"ipa": ipa, for d in first.get("defs", []):
"part_of_speech": pos, parts = d.split("\t", 1)
"definitions": defs[:3] defs.append(parts[1].strip() if len(parts) > 1 else parts[0].strip())
}
return {
"ipa": ipa,
"part_of_speech": pos,
"definitions": defs[:3]
}
except Exception: except Exception:
pass pass
return {"ipa": None, "part_of_speech": None, "definitions": []} return {"ipa": None, "part_of_speech": None, "definitions": []}
async def fetch_mymemory(query: str) -> Dict[str, Any]:
async def fetch_mymemory(query: str, client: Optional[httpx.AsyncClient] = None) -> Dict[str, Any]:
url = "https://api.mymemory.translated.net/get" url = "https://api.mymemory.translated.net/get"
params = { params = {
"q": query, "q": query,
@@ -129,45 +159,46 @@ async def fetch_mymemory(query: str) -> Dict[str, Any]:
"de": config.MYMEMORY_EMAIL "de": config.MYMEMORY_EMAIL
} }
try: try:
async with httpx.AsyncClient(timeout=4.0) as client: http = client or get_http_client()
resp = await client.get(url, params=params) resp = await http.get(url, params=params, timeout=4.0)
if resp.status_code == 200: if resp.status_code == 200:
data = resp.json() data = resp.json()
main_trans = data.get("responseData", {}).get("translatedText", "").strip() main_trans = data.get("responseData", {}).get("translatedText", "").strip()
matches = data.get("matches", []) matches = data.get("matches", [])
alt_list = [] alt_list = []
if main_trans: if main_trans:
alt_list.append(main_trans) alt_list.append(main_trans)
for m in matches: for m in matches:
t = m.get("translation", "").strip() t = m.get("translation", "").strip()
if t and t not in alt_list and t.lower() != query.lower(): if t and t not in alt_list and t.lower() != query.lower():
alt_list.append(t) alt_list.append(t)
return { return {
"translation": main_trans, "translation": main_trans,
"matches": alt_list[:6] "matches": alt_list[:6]
} }
except Exception: except Exception:
pass pass
return {"translation": "", "matches": []} return {"translation": "", "matches": []}
async def fetch_free_dictionary_api(word: str) -> Dict[str, Any]:
async def fetch_free_dictionary_api(word: str, client: Optional[httpx.AsyncClient] = None) -> Dict[str, Any]:
"""Optional tier: api.dictionaryapi.dev with fast timeout.""" """Optional tier: api.dictionaryapi.dev with fast timeout."""
url = f"https://api.dictionaryapi.dev/api/v2/entries/en/{urllib.parse.quote(word)}" url = f"https://api.dictionaryapi.dev/api/v2/entries/en/{urllib.parse.quote(word)}"
try: try:
async with httpx.AsyncClient(timeout=1.5) as client: http = client or get_http_client()
resp = await client.get(url) resp = await http.get(url, timeout=1.5)
if resp.status_code == 200: if resp.status_code == 200:
data = resp.json()[0] data = resp.json()[0]
phonetic = data.get("phonetic") phonetic = data.get("phonetic")
audio_url = None audio_url = None
for p in data.get("phonetics", []): for p in data.get("phonetics", []):
if p.get("audio"): if p.get("audio"):
audio_url = p.get("audio") audio_url = p.get("audio")
break break
return { return {
"phonetic": phonetic, "phonetic": phonetic,
"audio_url": audio_url "audio_url": audio_url
} }
except Exception: except Exception:
pass pass
return {"phonetic": None, "audio_url": None} return {"phonetic": None, "audio_url": None}
@@ -208,19 +239,21 @@ async def lookup_term(query: str) -> Dict[str, Any]:
# Parallel queries via asyncio # Parallel queries via asyncio
import asyncio import asyncio
http_client = get_http_client()
if is_sentence: if is_sentence:
# For long sentences, primary focus is translation # For long sentences, primary focus is translation
mm_data = await fetch_mymemory(query_clean) mm_data = await fetch_mymemory(query_clean, client=http_client)
result["translation"] = mm_data.get("translation", "") result["translation"] = mm_data.get("translation", "")
result["translations_list"] = mm_data.get("matches", []) result["translations_list"] = mm_data.get("matches", [])
else: else:
# For words and short phrases # For words and short phrases: reuse keep-alive connections across providers
tasks = [ tasks = [
fetch_wiktionary_wikitext(query_clean), fetch_wiktionary_wikitext(query_clean, client=http_client),
fetch_wiktionary_definitions(query_clean), fetch_wiktionary_definitions(query_clean, client=http_client),
fetch_datamuse(query_clean), fetch_datamuse(query_clean, client=http_client),
fetch_mymemory(query_clean), fetch_mymemory(query_clean, client=http_client),
fetch_free_dictionary_api(query_clean) fetch_free_dictionary_api(query_clean, client=http_client)
] ]
w_wiki, w_defs, d_muse, mm_data, free_dict = await asyncio.gather(*tasks) w_wiki, w_defs, d_muse, mm_data, free_dict = await asyncio.gather(*tasks)
+2 -1
View File
@@ -25,7 +25,7 @@ from app.auth import (
record_login_success, record_login_success,
change_user_password, change_user_password,
) )
from app.dictionary import lookup_term, USER_AGENT from app.dictionary import lookup_term, USER_AGENT, close_http_client
from app.export import export_csv, export_json, export_anki_tsv, export_markdown from app.export import export_csv, export_json, export_anki_tsv, export_markdown
from app.srs import calculate_sm2, generate_quiz_options, build_activity_heatmap, calculate_cefr_level from app.srs import calculate_sm2, generate_quiz_options, build_activity_heatmap, calculate_cefr_level
from app.reader import parse_epub, parse_text, parse_pdf, count_words from app.reader import parse_epub, parse_text, parse_pdf, count_words
@@ -36,6 +36,7 @@ async def lifespan(app: FastAPI):
init_db() init_db()
ensure_admin_user() ensure_admin_user()
yield yield
await close_http_client()
app = FastAPI( app = FastAPI(
+31
View File
@@ -1011,3 +1011,34 @@ def test_books_pdf_multipart_upload():
# Cleanup # Cleanup
client.delete(f"/api/books/{book_id}") client.delete(f"/api/books/{book_id}")
def test_http_connection_pooling():
import pytest
from app.dictionary import get_http_client, close_http_client, lookup_term
import asyncio
# 1. Singleton returns same instance
c1 = get_http_client()
c2 = get_http_client()
assert c1 is c2
assert not c1.is_closed
# 2. Check connection limits configuration
pool = c1._transport._pool
assert pool._max_keepalive_connections == 10
assert pool._max_connections == 20
# 3. Lookup using pooled client
res = asyncio.run(lookup_term("resilience"))
assert res["query"] == "resilience"
assert "translation" in res
# 4. Clean closure on shutdown
asyncio.run(close_http_client())
assert c1.is_closed
# Next call creates a new active client
c3 = get_http_client()
assert c3 is not c1
assert not c3.is_closed
asyncio.run(close_http_client())