Use the unified pydantic-settings configuration in run_llm

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-08-04 14:03:37 +08:00
co-authored by Claude Fable 5
parent b68393ab01
commit f9ab79f0c2
6 changed files with 70 additions and 107 deletions
+2
View File
@@ -3,5 +3,7 @@
# Authors: # Authors:
# imacat@mail.imacat.idv.tw (imacat), 2026/7/31 # imacat@mail.imacat.idv.tw (imacat), 2026/7/31
# The SQLAlchemy database URL.
SQLALCHEMY_DATABASE_URL="postgresql://user:password@host/db"
# The Anthropic API key # The Anthropic API key
ANTHROPIC_API_KEY=sk-ant-... ANTHROPIC_API_KEY=sk-ant-...
@@ -4,6 +4,14 @@ pop\_fem\_audit\_tools package
Submodules Submodules
---------- ----------
pop\_fem\_audit\_tools.config module
------------------------------------
.. automodule:: pop_fem_audit_tools.config
:members:
:show-inheritance:
:undoc-members:
pop\_fem\_audit\_tools.run\_llm module pop\_fem\_audit\_tools.run\_llm module
-------------------------------------- --------------------------------------
+2
View File
@@ -39,6 +39,8 @@ classifiers = [
"Topic :: Text Processing :: Linguistic", "Topic :: Text Processing :: Linguistic",
] ]
dependencies = [ dependencies = [
"pydantic-settings >= 2",
"SQLAlchemy >= 2",
"anthropic", "anthropic",
] ]
+48
View File
@@ -0,0 +1,48 @@
# Tools for A Feminist Audit of Pop Music.
# Copyright 2026 imacat. All rights reserved.
# Authors:
# imacat@mail.imacat.idv.tw (imacat), 2026/7/31
"""The configuration.
"""
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
"""The configuration settings."""
app_name: str = "Tools for A Feminist Audit of Pop Music"
"""The application name."""
admin_email: str = "imacat@mail.imacat.idv.tw"
"""The administrator email address."""
SQLALCHEMY_DATABASE_URL: str
"""The SQLAlchemy database URL."""
ANTHROPIC_API_KEY: str
"""The Anthropic API key."""
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
"""The model configuration."""
__settings: Settings | None = None
"""The configuration settings."""
def get_settings() -> Settings:
"""Returns the configuration settings.
:return: The configuration settings.
"""
global __settings
if __settings is None:
__settings = Settings()
return __settings
def set_settings(settings: Settings) -> None:
"""Sets the configuration settings.
:param settings: The configuration settings.
:return: None.
"""
global __settings
__settings = settings
+3 -50
View File
@@ -14,7 +14,6 @@ arbitration batch, and archives every artifact self-contained under
import argparse import argparse
import hashlib import hashlib
import json import json
import os
import sys import sys
import time import time
from datetime import datetime from datetime import datetime
@@ -23,6 +22,8 @@ from typing import Any
import anthropic import anthropic
from .config import get_settings
type Item = dict[str, str] type Item = dict[str, str]
"""An input item with "id" and "content".""" """An input item with "id" and "content"."""
@@ -135,49 +136,6 @@ def load_items(path: Path) -> list[Item]:
return items return items
def parse_env_file(path: Path) -> dict[str, str]:
"""Parse a simple KEY=VALUE .env file.
Blank lines and lines starting with "#" are ignored.
:param path: The path of the .env file.
:return: The key-value pairs; empty when the file is missing.
"""
values: dict[str, str] = {}
if not path.is_file():
return values
for line in path.read_text(encoding="utf-8").splitlines():
stripped: str = line.strip()
if stripped == "" or stripped.startswith("#"):
continue
if "=" not in stripped:
continue
key, _, value = stripped.partition("=")
values[key.strip()] = value.strip()
return values
def resolve_api_key(env_path: Path) -> str:
"""Resolve the Anthropic API key.
The ``ANTHROPIC_API_KEY`` environment variable takes precedence;
the .env file is consulted as a fallback.
:param env_path: The path of the .env file.
:return: The API key.
:raises RuntimeError: When no API key can be found.
"""
key: str | None = os.environ.get("ANTHROPIC_API_KEY")
if key:
return key
key = parse_env_file(env_path).get("ANTHROPIC_API_KEY")
if key:
return key
raise RuntimeError(
"ANTHROPIC_API_KEY is not set in the environment and not"
f" found in {env_path}")
def build_request(item: Item, system_prompt: str, def build_request(item: Item, system_prompt: str,
max_tokens: int) -> dict[str, Any]: max_tokens: int) -> dict[str, Any]:
"""Build one Message Batches request for an input item. """Build one Message Batches request for an input item.
@@ -571,13 +529,8 @@ def main(argv: list[str] | None = None) -> int:
print(f"dry run: archive created at {run_dir}", print(f"dry run: archive created at {run_dir}",
file=sys.stderr) file=sys.stderr)
return 0 return 0
try:
api_key: str = resolve_api_key(Path(".env"))
except RuntimeError as error:
print(f"error: {error}", file=sys.stderr)
return 1
client: anthropic.Anthropic = anthropic.Anthropic( client: anthropic.Anthropic = anthropic.Anthropic(
api_key=api_key) api_key=get_settings().ANTHROPIC_API_KEY)
run1: Results run1: Results
run2: Results run2: Results
run1, run2 = execute_runs( run1, run2 = execute_runs(
+7 -57
View File
@@ -15,7 +15,7 @@ from pathlib import Path
from typing import Any from typing import Any
from unittest import mock from unittest import mock
from pop_fem_audit_tools import run_llm from pop_fem_audit_tools import config, run_llm
class RunLLMTestCase(unittest.TestCase): class RunLLMTestCase(unittest.TestCase):
@@ -155,58 +155,6 @@ class TestLoadItems(RunLLMTestCase):
run_llm.load_items(path) run_llm.load_items(path)
class TestEnvParsing(RunLLMTestCase):
"""Test cases for the .env parsing and API key resolution."""
def setUp(self) -> None:
"""Create a temporary directory for the .env files."""
self.__dir: Path = self._make_temp_dir()
def test_parse_env_file(self) -> None:
"""Test parsing a .env file with comments and blanks."""
path: Path = self.__dir / ".env"
path.write_text(
"# a comment\n"
"\n"
"ANTHROPIC_API_KEY=sk-test-123\n"
"OTHER = value \n"
"garbage line\n",
encoding="utf-8")
values: dict[str, str] = run_llm.parse_env_file(path)
self.assertEqual(values,
{"ANTHROPIC_API_KEY": "sk-test-123",
"OTHER": "value"})
def test_parse_missing_env_file(self) -> None:
"""Test that a missing .env file yields no values."""
values: dict[str, str] = run_llm.parse_env_file(
self.__dir / ".env")
self.assertEqual(values, {})
def test_resolve_from_environment(self) -> None:
"""Test that the environment variable takes precedence."""
path: Path = self.__dir / ".env"
path.write_text("ANTHROPIC_API_KEY=sk-file\n",
encoding="utf-8")
with mock.patch.dict(os.environ,
{"ANTHROPIC_API_KEY": "sk-env"}):
self.assertEqual(run_llm.resolve_api_key(path), "sk-env")
def test_resolve_from_env_file(self) -> None:
"""Test that the .env file is used as a fallback."""
path: Path = self.__dir / ".env"
path.write_text("ANTHROPIC_API_KEY=sk-file\n",
encoding="utf-8")
with mock.patch.dict(os.environ, {}, clear=True):
self.assertEqual(run_llm.resolve_api_key(path), "sk-file")
def test_resolve_missing_key(self) -> None:
"""Test that a missing API key raises an error."""
with mock.patch.dict(os.environ, {}, clear=True):
with self.assertRaises(RuntimeError):
run_llm.resolve_api_key(self.__dir / ".env")
class TestRequestBuilding(RunLLMTestCase): class TestRequestBuilding(RunLLMTestCase):
"""Test cases for the request construction.""" """Test cases for the request construction."""
@@ -373,6 +321,10 @@ class TestMainFlow(RunLLMTestCase):
"--arbitration-prompt", "prompts/task_arbitration_v1.md", "--arbitration-prompt", "prompts/task_arbitration_v1.md",
"--input", "items.jsonl", "--input", "items.jsonl",
"--phase", "coding"] "--phase", "coding"]
self.__settings: config.Settings = config.Settings(
SQLALCHEMY_DATABASE_URL="sqlite://",
ANTHROPIC_API_KEY="test-key")
config.set_settings(self.__settings)
@staticmethod @staticmethod
def __make_client(run1: list[Any], run2: list[Any], def __make_client(run1: list[Any], run2: list[Any],
@@ -414,10 +366,8 @@ class TestMainFlow(RunLLMTestCase):
""" """
stdout: io.StringIO = io.StringIO() stdout: io.StringIO = io.StringIO()
stderr: io.StringIO = io.StringIO() stderr: io.StringIO = io.StringIO()
environ: dict[str, str] = {"ANTHROPIC_API_KEY": "sk-test"} with mock.patch.object(run_llm.anthropic, "Anthropic",
with mock.patch.dict(os.environ, environ), \ return_value=client), \
mock.patch.object(run_llm.anthropic, "Anthropic",
return_value=client), \
redirect_stdout(stdout), redirect_stderr(stderr): redirect_stdout(stdout), redirect_stderr(stderr):
status: int = run_llm.main(argv) status: int = run_llm.main(argv)
return status, stdout.getvalue() return status, stdout.getvalue()