feat(advice): LLM brewing advice on every observation #14
@@ -46,6 +46,33 @@ class BeerRepository:
|
||||
self.run_meta = self.db["beer_runs"]
|
||||
self._ensure_indexes()
|
||||
|
||||
def close(self) -> None:
|
||||
"""Close the underlying Mongo client and release it.
|
||||
|
||||
A ``MongoClient`` keeps a connection pool and spawns three
|
||||
monitoring threads. ``close()`` alone does NOT reap them - they
|
||||
go when the client object is garbage collected - so the
|
||||
reference is dropped as well.
|
||||
|
||||
The request-scoped repository is closed by the app's teardown, but
|
||||
a caller that builds its own - the advice background thread, for
|
||||
instance - has to close it itself or the threads and the pool
|
||||
accumulate, three per reading.
|
||||
"""
|
||||
client = self.client
|
||||
try:
|
||||
client.close()
|
||||
except Exception:
|
||||
pass
|
||||
# Drop the reference so the monitor threads can be reaped.
|
||||
self.client = None
|
||||
|
||||
def __enter__(self) -> "BeerRepository":
|
||||
return self
|
||||
|
||||
def __exit__(self, *exc_info) -> None:
|
||||
self.close()
|
||||
|
||||
def _ensure_indexes(self) -> None:
|
||||
"""Make sure the lookup keys are indexed.
|
||||
|
||||
@@ -205,8 +232,10 @@ class BeerRepository:
|
||||
doc_id = _as_doc_id(reading_id)
|
||||
if doc_id is None:
|
||||
return 0
|
||||
# `$in` matches either form. A plain list would be an exact
|
||||
# ARRAY match in Mongo, which is not what is wanted here.
|
||||
result = self.readings.update_one(
|
||||
{"_id": doc_id},
|
||||
{"_id": {"$in": doc_id}},
|
||||
{"$set": {
|
||||
"advice": str(advice),
|
||||
"advice_ok": bool(ok),
|
||||
@@ -341,12 +370,42 @@ def _as_id(value: object) -> str:
|
||||
return ""
|
||||
|
||||
|
||||
#: A process-wide repository, for the code that runs outside a request.
|
||||
#:
|
||||
#: A ``MongoClient`` spawns three monitor threads and pymongo does not
|
||||
#: reap them on ``close()`` - they go only when the client is garbage
|
||||
#: collected. So building a client per advice write accumulated three
|
||||
#: threads and a connection pool for every reading recorded. One shared
|
||||
#: client for the life of the process is what the request-scoped
|
||||
#: repository does too.
|
||||
_shared_repository: "BeerRepository | None" = None
|
||||
|
||||
|
||||
def get_shared_repository() -> "BeerRepository":
|
||||
"""The process-wide repository, built on first use."""
|
||||
global _shared_repository
|
||||
|
||||
if _shared_repository is None:
|
||||
_shared_repository = BeerRepository()
|
||||
return _shared_repository
|
||||
|
||||
|
||||
def _as_doc_id(value: object):
|
||||
"""Coerce a reading document id to something Mongo will match.
|
||||
|
||||
The ``_id`` of a reading is an ``ObjectId``, not a run id, so the
|
||||
``_as_id`` coercion does not apply - it would turn the ObjectId into
|
||||
an empty string and the update would silently match nothing.
|
||||
The ``_id`` of a reading is not a run id, so the ``_as_id`` coercion
|
||||
does not apply - it would turn the id into an empty string and the
|
||||
update would silently match nothing.
|
||||
|
||||
Two shapes exist in this database:
|
||||
|
||||
- the ORIGINAL app's ids are 24 random CHARACTERS stored as a string
|
||||
(``l78ZidBJ3QaYn98tvB88x4uM``), and
|
||||
- ids written by Mongo itself are ``ObjectId`` objects.
|
||||
|
||||
So the id is returned in BOTH forms and the query matches either.
|
||||
Accepting only the ObjectId form would mean the advice was never
|
||||
stored for any of the existing readings.
|
||||
|
||||
Only an ObjectId or a plain string is accepted, so a dict can never
|
||||
be injected as a query operator.
|
||||
@@ -354,12 +413,15 @@ def _as_doc_id(value: object):
|
||||
from bson import ObjectId
|
||||
|
||||
if isinstance(value, ObjectId):
|
||||
return value
|
||||
if isinstance(value, str):
|
||||
return [value]
|
||||
if isinstance(value, str) and value:
|
||||
try:
|
||||
return ObjectId(value)
|
||||
# A 24-character hex string may or may not be the stored
|
||||
# ObjectId or the stored string, so try both.
|
||||
return [ObjectId(value), value]
|
||||
except Exception:
|
||||
return None
|
||||
# Not hex - it can only be the original app's string id.
|
||||
return [value]
|
||||
return None
|
||||
|
||||
|
||||
|
||||
@@ -89,7 +89,7 @@ def _start_advice(run_id: str, reading_id: str) -> None:
|
||||
for reading in run.readings:
|
||||
points = reading.sg_points
|
||||
readings.append({
|
||||
"_id": reading._id,
|
||||
"_id": str(reading._id),
|
||||
"date": reading.date.isoformat() if reading.date else "",
|
||||
"sg": points,
|
||||
"sg_decimal": reading.sg_decimal,
|
||||
@@ -98,7 +98,14 @@ def _start_advice(run_id: str, reading_id: str) -> None:
|
||||
"final": bool(reading.final_reading),
|
||||
})
|
||||
|
||||
latest = dict(readings[-1]) if readings else {"date": "", "sg": None}
|
||||
# The reading that is actually being advised about. It is not
|
||||
# necessarily the newest one: the brewer may be editing an older
|
||||
# reading, and advice about the newest reading would be about the
|
||||
# wrong number.
|
||||
saved = next((r for r in readings if r.get("_id") == str(reading_id)), None)
|
||||
if saved is None:
|
||||
saved = dict(readings[-1]) if readings else {"date": "", "sg": None}
|
||||
latest = dict(saved)
|
||||
latest["_id"] = reading_id
|
||||
|
||||
except Exception: # noqa: BLE001 - advice must never break the save
|
||||
@@ -106,8 +113,21 @@ def _start_advice(run_id: str, reading_id: str) -> None:
|
||||
return
|
||||
|
||||
def write(read_id: str, ok: bool, text: str) -> None:
|
||||
from mongo.repository import BeerRepository
|
||||
BeerRepository().set_reading_advice(read_id, text, ok)
|
||||
# The client is SHARED, not built per reading. A MongoClient
|
||||
# spawns three monitor threads and pymongo does not reap them on
|
||||
# close(), so building one per advice write accumulated three
|
||||
# threads and a connection pool for every reading recorded.
|
||||
#
|
||||
# The shared client is created on first use and lives for the
|
||||
# life of the process, which is what the request-scoped
|
||||
# repository does too.
|
||||
from mongo.repository import BeerRepository, get_shared_repository
|
||||
|
||||
try:
|
||||
shared = get_shared_repository()
|
||||
shared.set_reading_advice(read_id, text, ok)
|
||||
except Exception: # noqa: BLE001 - never let the thread die noisily
|
||||
log.exception("could not store the generated advice")
|
||||
|
||||
thread = threading.Thread(
|
||||
target=brew_advice.generate_advice_in_background,
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from contextlib import contextmanager
|
||||
from datetime import date
|
||||
|
||||
import pytest
|
||||
@@ -248,6 +249,153 @@ def test_editing_a_reading_clears_its_stale_advice(repo):
|
||||
assert after["sg"] == "1030"
|
||||
|
||||
|
||||
def test_the_advice_is_stored_on_a_reading_with_a_string_id(repo):
|
||||
"""The original app's readings have 24-character STRING ids.
|
||||
|
||||
This is the shape every existing reading in the live database has -
|
||||
the ids were generated by the old app as random characters, not by
|
||||
Mongo. A helper that only accepted an ObjectId would have matched
|
||||
nothing, and the advice would silently never be stored.
|
||||
|
||||
The mongomock fixture generates ObjectIds, so this has to be
|
||||
constructed explicitly or the bug is invisible in the tests.
|
||||
"""
|
||||
repo.readings.insert_one({
|
||||
"_id": "l78ZidBJ3QaYn98tvB88x4uM",
|
||||
"beer_run_id": "1",
|
||||
"beer_type": "Ale",
|
||||
"sg": "1042",
|
||||
"date": "2026-01-01",
|
||||
"final_reading": "False",
|
||||
"comment": "First ever try",
|
||||
})
|
||||
|
||||
modified = repo.set_reading_advice("l78ZidBJ3QaYn98tvB88x4uM",
|
||||
"THE ADVICE", True)
|
||||
assert modified == 1, "the advice was not stored on a string-id reading"
|
||||
|
||||
after = repo.readings.find_one({"_id": "l78ZidBJ3QaYn98tvB88x4uM"})
|
||||
assert after["advice"] == "THE ADVICE"
|
||||
assert after["comment"] == "First ever try"
|
||||
|
||||
|
||||
def test_the_advice_is_stored_on_a_reading_with_an_object_id(repo):
|
||||
"""The other shape: a reading written by Mongo itself."""
|
||||
seed_readings(repo, [("1", "Ale", "1042", "2026-01-01", False, "a comment")])
|
||||
doc = repo.readings.find_one({"beer_run_id": "1"})
|
||||
|
||||
modified = repo.set_reading_advice(str(doc["_id"]), "THE ADVICE", True)
|
||||
assert modified == 1
|
||||
|
||||
after = repo.readings.find_one({"beer_run_id": "1"})
|
||||
assert after["advice"] == "THE ADVICE"
|
||||
|
||||
|
||||
def test_writing_the_advice_does_not_leak_a_client_per_reading():
|
||||
"""A MongoClient spawns three monitor threads and pymongo does not
|
||||
reap them on close() - they go only when the client is collected.
|
||||
|
||||
So building a client per advice write accumulated three threads and a
|
||||
connection pool for every reading recorded. The advice writer shares
|
||||
one client for the life of the process instead.
|
||||
"""
|
||||
import inspect
|
||||
|
||||
from views import brews as brews_module
|
||||
|
||||
# The writer must NOT build a repository of its own - it has to use
|
||||
# the shared one. A MongoClient spawns three monitor threads and
|
||||
# pymongo does not reap them on close(), so a client per advice write
|
||||
# accumulated three threads and a connection pool for every reading
|
||||
# recorded.
|
||||
source = inspect.getsource(brews_module._start_advice)
|
||||
|
||||
assert "BeerRepository()" not in source, (
|
||||
"the advice writer builds its own repository - that leaks a Mongo "
|
||||
"client (three monitor threads) per reading"
|
||||
)
|
||||
assert "get_shared_repository" in source, (
|
||||
"the advice writer must use the shared repository"
|
||||
)
|
||||
|
||||
|
||||
@contextmanager
|
||||
def _app_with_repo(repo):
|
||||
"""An application context whose repository is *repo*.
|
||||
|
||||
The view reads the repository from ``app.extensions``, so pointing
|
||||
that at a fixture is enough to drive the view's code directly.
|
||||
"""
|
||||
from app_factory import create_app
|
||||
|
||||
app = create_app({
|
||||
"TESTING": True,
|
||||
"WTF_CSRF_ENABLED": False,
|
||||
"SECRET_KEY": "test-secret",
|
||||
"MONGO_CLIENT": repo.client,
|
||||
})
|
||||
# The advice must be generated even though this is a test app.
|
||||
app.config["TESTING"] = False
|
||||
|
||||
original = app.extensions["beer_repo"]
|
||||
app.extensions["beer_repo"] = lambda: repo
|
||||
with app.app_context():
|
||||
try:
|
||||
yield app
|
||||
finally:
|
||||
app.extensions["beer_repo"] = original
|
||||
|
||||
|
||||
def test_the_advice_is_about_the_reading_that_was_saved():
|
||||
"""Editing an OLDER reading must not generate advice about the newest.
|
||||
|
||||
The advice is written against a specific reading, so the reading being
|
||||
saved has to be the one the advice is about.
|
||||
"""
|
||||
import time
|
||||
|
||||
import mongomock
|
||||
from mongo.repository import BeerRepository
|
||||
from views import brews as brews_module
|
||||
from views.brews import _start_advice
|
||||
|
||||
captured = {}
|
||||
|
||||
def fake_generate(run, reading, history, writer=None):
|
||||
captured["reading"] = reading
|
||||
captured["history"] = history
|
||||
return True, "THE ADVICE"
|
||||
|
||||
original = brews_module.brew_advice.generate_advice_in_background
|
||||
brews_module.brew_advice.generate_advice_in_background = fake_generate
|
||||
try:
|
||||
# Two readings; the OLDER one is the one being saved.
|
||||
repo = BeerRepository(client=mongomock.MongoClient())
|
||||
old_id = repo.upsert_reading(beer_run_id="1", beer_type="Ale", sg=1042,
|
||||
reading_date="2026-01-01", final_reading=False,
|
||||
comment="first")
|
||||
repo.upsert_reading(beer_run_id="1", beer_type="Ale", sg=1030,
|
||||
reading_date="2026-01-10", final_reading=False,
|
||||
comment="second")
|
||||
|
||||
# The advice is only generated when it is enabled, and it is off
|
||||
# in testing.
|
||||
with _app_with_repo(repo):
|
||||
_start_advice("1", old_id)
|
||||
for _ in range(40):
|
||||
time.sleep(0.05)
|
||||
if captured:
|
||||
break
|
||||
finally:
|
||||
brews_module.brew_advice.generate_advice_in_background = original
|
||||
|
||||
assert captured, "the advice was never generated"
|
||||
assert captured["reading"]["_id"] == old_id
|
||||
assert captured["reading"]["sg"] == 1042, (
|
||||
"the advice was generated about the wrong reading"
|
||||
)
|
||||
|
||||
|
||||
def test_the_reading_id_cannot_be_used_to_inject_a_query(repo):
|
||||
"""The reading id is coerced before it reaches Mongo.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user