Package updates

pyproject.toml

@@ -23,11 +23,11 @@ dependencies = [
     "ldap3",
     "orjson",
     "pamela",
-    "pydantic",
-    "pydantic-settings",
+    "pydantic >=2, <3",
+    "pydantic-settings >=2",
     "python-jose",
     "pyzmq",
-    "sqlalchemy",
+    "sqlalchemy >=2",
     "starlette",
     "uvicorn",
 ]

src/bluesky_httpserver/app.py

@@ -272,9 +272,10 @@ async def startup_event():
             async def purge_expired_sessions_and_api_keys():
                 logger.info("Purging expired Sessions and API keys from the database.")
                 while True:
-                    await asyncio.get_running_loop().run_in_executor(None, purge_expired(engine, orm.Session))
-                    await asyncio.get_running_loop().run_in_executor(None, purge_expired(engine, orm.APIKey))
+                    await asyncio.get_running_loop().run_in_executor(None, purge_expired, engine, orm.Session)
+                    await asyncio.get_running_loop().run_in_executor(None, purge_expired, engine, orm.APIKey)
                     await asyncio.sleep(600)
+                    engine.dispose()
 
             app.state.tasks.append(asyncio.create_task(purge_expired_sessions_and_api_keys()))
 

src/bluesky_httpserver/database/base.py

@@ -1,5 +1,7 @@
-from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import DeclarativeBase
+
 
 # Everything imports this so we put it in its own module to
 # avoid circular imports.
-Base = declarative_base()
+class Base(DeclarativeBase):
+    pass

src/bluesky_httpserver/database/core.py

@@ -183,13 +183,16 @@ def purge_expired(engine, cls):
     """
     SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
     db = SessionLocal()
-    now = datetime.utcnow()
-    deleted = False
-    for obj in db.query(cls).filter(cls.expiration_time.is_not(None)).filter(cls.expiration_time < now):
-        deleted = True
-        db.delete(obj)
-    if deleted:
-        db.commit()
+    try:
+        now = datetime.utcnow()
+        deleted = False
+        for obj in db.query(cls).filter(cls.expiration_time.is_not(None)).filter(cls.expiration_time < now):
+            deleted = True
+            db.delete(obj)
+        if deleted:
+            db.commit()
+    finally:
+        db.close()
     return cls
 
 

src/bluesky_httpserver/schemas.py

@@ -33,9 +33,11 @@ class Response(generic_model, Generic[DataT, LinksT, MetaT]):
     links: LinksT | None = None
     meta: MetaT | None = None
 
-    @pydantic.validator("error", always=True)
-    def check_consistency(cls, v, values):
-        if v is not None and values["data"] is not None:
+    @pydantic.field_validator("error", mode="after")
+    @classmethod
+    def check_consistency(cls, v, info):
+        values = info.data
+        if v is not None and values.get("data") is not None:
             raise ValueError("must not provide both data and error")
         if v is None and values.get("data") is None:
             raise ValueError("must provide data or error")
@@ -310,7 +312,7 @@ class APIKeyRequestParams(pydantic.BaseModel):
     # Provide an example for expires_in. Otherwise, OpenAPI suggests lifetime=0.
     # If the user is not reading carefully, they will be frustrated when they
     # try to use the instantly-expiring API key!
-    expires_in: int | None = pydantic.Field(..., example=600)  # seconds
-    # scopes: Optional[List[str]] = pydantic.Field(..., example=["inherit"])
-    scopes: list[str] | None = pydantic.Field(default=["inherit"], example=["inherit"])
+    expires_in: int | None = pydantic.Field(..., json_schema_extra={"example": 600})  # seconds
+    # scopes: Optional[List[str]] = pydantic.Field(..., json_schema_extra={"example": ["inherit"]})
+    scopes: list[str] | None = pydantic.Field(default=["inherit"], json_schema_extra={"example": ["inherit"]})
     note: str | None = None

src/bluesky_httpserver/settings.py

@@ -60,7 +60,7 @@ def database_settings(self):
         )
 
 
-@lru_cache
+@lru_cache(1)
 def get_settings():
     return Settings()
 

src/bluesky_httpserver/tests/conftest.py

@@ -47,6 +47,7 @@ def _wait_for_http_server_ready(*, timeout=10, request_prefix="/api"):
 
 @pytest.fixture(scope="module")
 def fastapi_server(xprocess):
+
     class Starter(ProcessStarter):
         env = dict(os.environ)
         env["QSERVER_HTTP_SERVER_SINGLE_USER_API_KEY"] = API_KEY_FOR_TESTS
@@ -298,3 +299,25 @@ def well_known_response(oidc_base_url: str) -> dict:
         "device_authorization_endpoint": f"{oidc_base_url}protocol/openid-connect/auth/device",
         "end_session_endpoint": f"{oidc_base_url}protocol/openid-connect/logout",
     }
+
+
+@pytest.fixture(scope="session", autouse=True)
+def print_open_file_descriptors(request):
+    yield
+    ttime.sleep(1)
+    pid = os.getpid()
+    fd_dir = f"/proc/{pid}/fd"
+    fd_entries = sorted(os.listdir(fd_dir), key=int)
+    msg = f"+++ PID={pid} OPEN FILE DESCRIPTORS = {len(fd_entries)}"
+    # terminalreporter works on CI (it is supposed to work locally, but it doesn't)
+    reporter = request.config.pluginmanager.get_plugin("terminalreporter")
+    reporter.write_line(msg)
+    # /dev/tty bypasses all pytest capture layers (local use only)
+    try:
+        with open("/dev/tty", "w") as tty:
+            tty.write("\n" + msg + "\n")
+    except OSError:
+        import sys
+
+        sys.stderr.write("\n" + msg + "\n")
+        sys.stderr.flush()

src/bluesky_httpserver/tests/test_core_api_fs.py

@@ -96,48 +96,50 @@ def test_http_server_queue_upload_spreasheet_1(
     ss_path, plans_expected = _create_test_excel_file1(tmp_path, plan_params=plan_params, col_names=col_names)
 
     # Send the Excel file to the server
-    files = {"spreadsheet": open(ss_path, "rb")}
-    resp1 = request_to_json("post", "/queue/upload/spreadsheet", files=files)
-    assert "success" in resp1, str(resp1)
-    assert resp1["success"] is True, str(resp1)
-    items1 = resp1["items"]
-    results1 = resp1["results"]
-    assert len(items1) == len(plans_expected), str(items1)
-    for p, p_exp in zip(items1, plans_expected):
-        for k, v in p_exp.items():
-            assert k in p
-            assert v == p[k]
-
-    assert len(results1) == len(plans_expected), str(results1)
-    assert all(_["success"] is True for _ in results1), str(results1)
-    assert all(_["msg"] == "" for _ in results1), str(results1)
+    with open(ss_path, "rb") as f:
+        files = {"spreadsheet": f}
+        resp1 = request_to_json("post", "/queue/upload/spreadsheet", files=files)
+        assert "success" in resp1, str(resp1)
+        assert resp1["success"] is True, str(resp1)
+        items1 = resp1["items"]
+        results1 = resp1["results"]
+        assert len(items1) == len(plans_expected), str(items1)
+        for p, p_exp in zip(items1, plans_expected):
+            for k, v in p_exp.items():
+                assert k in p
+                assert v == p[k]
+
+        assert len(results1) == len(plans_expected), str(results1)
+        assert all(_["success"] is True for _ in results1), str(results1)
+        assert all(_["msg"] == "" for _ in results1), str(results1)
+
+        # Verify that the queue contains correct plans
+        resp2 = request_to_json("get", "/queue/get")
+        assert resp2["success"] is True
+        assert resp2["running_item"] == {}
+        queue = resp2["items"]
+        assert len(queue) == len(plans_expected), str(queue)
+        for p, p_exp in zip(queue, plans_expected):
+            for k, v in p_exp.items():
+                assert k in p
+                assert v == p[k]
+
+        resp3 = request_to_json("post", "/environment/open")
+        assert resp3["success"] is True
+        assert wait_for_environment_to_be_created(10)
+
+        resp4 = request_to_json("post", "/queue/start")
+        assert resp4["success"] is True
+        assert wait_for_queue_execution_to_complete(60)
+
+        resp5 = request_to_json("get", "/status")
+        assert resp5["items_in_queue"] == 0
+        assert resp5["items_in_history"] == len(plans_expected)
+
+        resp6 = request_to_json("post", "/environment/close")
+        assert resp6 == {"success": True, "msg": ""}
+        assert wait_for_manager_state_idle(10)
 
-    # Verify that the queue contains correct plans
-    resp2 = request_to_json("get", "/queue/get")
-    assert resp2["success"] is True
-    assert resp2["running_item"] == {}
-    queue = resp2["items"]
-    assert len(queue) == len(plans_expected), str(queue)
-    for p, p_exp in zip(queue, plans_expected):
-        for k, v in p_exp.items():
-            assert k in p
-            assert v == p[k]
-
-    resp3 = request_to_json("post", "/environment/open")
-    assert resp3["success"] is True
-    assert wait_for_environment_to_be_created(10)
-
-    resp4 = request_to_json("post", "/queue/start")
-    assert resp4["success"] is True
-    assert wait_for_queue_execution_to_complete(60)
-
-    resp5 = request_to_json("get", "/status")
-    assert resp5["items_in_queue"] == 0
-    assert resp5["items_in_history"] == len(plans_expected)
-
-    resp6 = request_to_json("post", "/environment/close")
-    assert resp6 == {"success": True, "msg": ""}
-    assert wait_for_manager_state_idle(10)
 
 
 def test_http_server_queue_upload_spreasheet_2(re_manager, fastapi_server_fs, tmp_path, monkeypatch):  # noqa F811
@@ -158,13 +160,14 @@ def test_http_server_queue_upload_spreasheet_2(re_manager, fastapi_server_fs, tm
     ss_path, plans_expected = _create_test_excel_file1(tmp_path, plan_params=plan_params, col_names=col_names)
 
     # Send the Excel file to the server
-    files = {"spreadsheet": open(ss_path, "rb")}
-    data = {"data_type": "unsupported"}
-    resp1 = request_to_json("post", "/queue/upload/spreadsheet", files=files, data=data)
-    assert resp1["success"] is False, str(resp1)
-    assert resp1["msg"] == "Unsupported data type: 'unsupported'"
-    assert resp1["items"] == []
-    assert resp1["results"] == []
+    with open(ss_path, "rb") as f:
+        files = {"spreadsheet": f}
+        data = {"data_type": "unsupported"}
+        resp1 = request_to_json("post", "/queue/upload/spreadsheet", files=files, data=data)
+        assert resp1["success"] is False, str(resp1)
+        assert resp1["msg"] == "Unsupported data type: 'unsupported'"
+        assert resp1["items"] == []
+        assert resp1["results"] == []
 
 
 def test_http_server_queue_upload_spreasheet_3(re_manager, fastapi_server_fs, tmp_path, monkeypatch):  # noqa F811
@@ -189,10 +192,11 @@ def test_http_server_queue_upload_spreasheet_3(re_manager, fastapi_server_fs, tm
     os.rename(ss_path, new_path)
 
     # Send the Excel file to the server
-    files = {"spreadsheet": open(new_path, "rb")}
-    resp1 = request_to_json("post", "/queue/upload/spreadsheet", files=files)
-    assert resp1["success"] is False, str(resp1)
-    assert resp1["msg"] == f"Unsupported file (extension '{new_ext}')"
+    with open(new_path, "rb") as f:
+        files = {"spreadsheet": f}
+        resp1 = request_to_json("post", "/queue/upload/spreadsheet", files=files)
+        assert resp1["success"] is False, str(resp1)
+        assert resp1["msg"] == f"Unsupported file (extension '{new_ext}')"
 
 
 @pytest.mark.parametrize("use_custom", [False, True])
@@ -275,27 +279,29 @@ def test_http_server_queue_upload_spreasheet_5(re_manager, fastapi_server_fs, tm
     ss_path, plans_expected = _create_test_excel_file1(tmp_path, plan_params=plan_params, col_names=col_names)
 
     # Send the Excel file to the server
-    files = {"spreadsheet": open(ss_path, "rb")}
-    resp1 = request_to_json("post", "/queue/upload/spreadsheet", files=files)
-    assert resp1["success"] is False, str(resp1)
-    assert resp1["msg"] == "Failed to add all items: validation of 1 out of 3 submitted items failed"
-
-    items, results = resp1["items"], resp1["results"]
-    assert len(items) == len(plans_expected), str(items)
-    assert len(results) == len(plans_expected), str(items)
-    for n, p_exp in enumerate(plans_expected):
-        p, r = items[n], results[n]
-        if p_exp["name"] == "nonexisting_plan":
-            assert r["success"] is False
-            assert "not in the list of allowed plans" in r["msg"], r["msg"]
-        else:
-            assert r["success"] is True
-            assert r["msg"] == ""
-        for k, v in p_exp.items():
-            assert k in p
-            assert v == p[k]
-
-    # No plans are expected to be added to the queue
-    resp2 = request_to_json("get", "/status")
-    assert resp2["items_in_queue"] == 0
-    assert resp2["items_in_history"] == 0
+    with open(ss_path, "rb") as f:
+        files = {"spreadsheet": f}
+
+        resp1 = request_to_json("post", "/queue/upload/spreadsheet", files=files)
+        assert resp1["success"] is False, str(resp1)
+        assert resp1["msg"] == "Failed to add all items: validation of 1 out of 3 submitted items failed"
+
+        items, results = resp1["items"], resp1["results"]
+        assert len(items) == len(plans_expected), str(items)
+        assert len(results) == len(plans_expected), str(items)
+        for n, p_exp in enumerate(plans_expected):
+            p, r = items[n], results[n]
+            if p_exp["name"] == "nonexisting_plan":
+                assert r["success"] is False
+                assert "not in the list of allowed plans" in r["msg"], r["msg"]
+            else:
+                assert r["success"] is True
+                assert r["msg"] == ""
+            for k, v in p_exp.items():
+                assert k in p
+                assert v == p[k]
+
+        # No plans are expected to be added to the queue
+        resp2 = request_to_json("get", "/status")
+        assert resp2["items_in_queue"] == 0
+        assert resp2["items_in_history"] == 0