4 Commits
11 changed files with 44 additions and 34 deletions
+2 -2
View File
@@ -3,7 +3,7 @@
# Authors:
# imacat@mail.imacat.idv.tw (imacat), 2026/7/31
# The SQLAlchemy database URL.
SQLALCHEMY_DATABASE_URL="postgresql://user:password@host/db"
# The SQLAlchemy database URI.
SQLALCHEMY_DATABASE_URI="postgresql://user:password@host/db"
# The Anthropic API key
ANTHROPIC_API_KEY=sk-ant-...
+2 -2
View File
@@ -14,12 +14,12 @@ class Settings(BaseSettings):
"""The application name."""
admin_email: str = "imacat@mail.imacat.idv.tw"
"""The administrator email address."""
SQLALCHEMY_DATABASE_URL: str
SQLALCHEMY_DATABASE_URI: str
"""The SQLAlchemy database URL."""
ANTHROPIC_API_KEY: str
"""The Anthropic API key."""
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
model_config = SettingsConfigDict(env_file=".env")
"""The model configuration."""
+30 -21
View File
@@ -7,10 +7,11 @@
"""
from functools import cached_property
from pathlib import Path
from typing import Any
import sqlalchemy as sa
from sqlalchemy.engine.interfaces import DBAPICursor, DBAPIConnection
from sqlalchemy.orm import DeclarativeBase, sessionmaker, Session
from sqlalchemy.pool import ConnectionPoolEntry
from .config import Settings, get_settings
@@ -29,7 +30,7 @@ class DataSource:
:return: The database engine.
"""
settings: Settings = get_settings()
return self.__create_engine(settings.SQLALCHEMY_DATABASE_URL)
return self.__create_engine(settings.SQLALCHEMY_DATABASE_URI)
@cached_property
def __session_local(self) -> sessionmaker:
@@ -51,9 +52,6 @@ class DataSource:
def __create_engine(cls, url: str) -> sa.Engine:
"""Constructs and returns the database engine.
The foreign key enforcement is enabled on every connection
of a SQLite engine.
:param url: The SQLAlchemy database URL.
:return: The database engine.
"""
@@ -65,24 +63,10 @@ class DataSource:
poolclass=sa.StaticPool)
else:
engine = sa.create_engine(url)
if engine.url.get_backend_name() == "sqlite":
sa.event.listen(engine, "connect",
cls.__enable_sqlite_foreign_keys)
if engine.dialect.name == "sqlite":
cls.__enable_sqlite_foreign_keys(engine)
return engine
@staticmethod
def __enable_sqlite_foreign_keys(dbapi_connection: Any,
_: Any) -> None:
"""Enables the foreign key enforcement on a new connection.
:param dbapi_connection: The DBAPI connection.
:param _: The connection record (unused).
:return: None.
"""
cursor: Any = dbapi_connection.cursor()
cursor.execute("PRAGMA foreign_keys=ON")
cursor.close()
@staticmethod
def __resolve_sqlite_relative_url(url: str) -> str:
"""Resolves the SQLite relative URL to the instance folder.
@@ -101,6 +85,31 @@ class DataSource:
path = base / "instance" / path
return f"sqlite:///{path}"
@staticmethod
def __enable_sqlite_foreign_keys(engine: sa.Engine) -> None:
"""Turns on the foreign key enforcement of SQLite.
The ``foreign_keys`` pragma is turned on for every
connection of the engine, so that the ``ON DELETE``
actions of the schema run.
:param engine: The SQLite database engine.
:return: None.
"""
def on_connect(dbapi_connection: DBAPIConnection,
_: ConnectionPoolEntry) -> None:
"""Turns on the pragma on a new connection.
:param dbapi_connection: The DB-API connection.
:param _: The connection record (unused).
:return: None.
"""
cursor: DBAPICursor = dbapi_connection.cursor()
cursor.execute("PRAGMA foreign_keys=ON")
cursor.close()
sa.event.listen(engine, "connect", on_connect)
ds: DataSource = DataSource()
"""The data source."""
+1 -1
View File
@@ -278,7 +278,7 @@ class TestBuildDB(unittest.TestCase):
self.__annotations: Path = self.__dir / "annotations.csv"
self.__write_chart(self.CHART_CSV)
config.set_settings(config.Settings(
SQLALCHEMY_DATABASE_URL="sqlite://",
SQLALCHEMY_DATABASE_URI="sqlite://",
ANTHROPIC_API_KEY="test-key"))
self.__ds: DataSource = DataSource()
self.addCleanup(self.__ds.engine.dispose)
+1 -1
View File
@@ -36,7 +36,7 @@ class TestExportLlmInput(unittest.TestCase):
self.__dir: Path = Path(tmp.name)
self.__output: Path = self.__dir / "llm-input.jsonl"
config.set_settings(config.Settings(
SQLALCHEMY_DATABASE_URL="sqlite://",
SQLALCHEMY_DATABASE_URI="sqlite://",
ANTHROPIC_API_KEY="test-key"))
self.__ds: DataSource = DataSource()
self.addCleanup(self.__ds.engine.dispose)
+1 -1
View File
@@ -45,7 +45,7 @@ class TestFetchArtists(unittest.TestCase):
self.__snapshot: Path = \
self.__dir / "artists_wikidata.csv"
config.set_settings(config.Settings(
SQLALCHEMY_DATABASE_URL="sqlite://",
SQLALCHEMY_DATABASE_URI="sqlite://",
ANTHROPIC_API_KEY="test-key"))
self.__ds: DataSource = DataSource()
self.addCleanup(self.__ds.engine.dispose)
+1 -1
View File
@@ -44,7 +44,7 @@ class TestFetchLyrics(unittest.TestCase):
self.__provenance: Path = \
self.__dir / "lyrics-provenance.csv"
config.set_settings(config.Settings(
SQLALCHEMY_DATABASE_URL="sqlite://",
SQLALCHEMY_DATABASE_URI="sqlite://",
ANTHROPIC_API_KEY="test-key"))
self.__ds: DataSource = DataSource()
self.addCleanup(self.__ds.engine.dispose)
+1 -1
View File
@@ -27,7 +27,7 @@ class TestModels(unittest.TestCase):
def setUp(self) -> None:
"""Create the schema on an in-memory SQLite database."""
config.set_settings(config.Settings(
SQLALCHEMY_DATABASE_URL="sqlite://",
SQLALCHEMY_DATABASE_URI="sqlite://",
ANTHROPIC_API_KEY="test-key"))
self.__ds: DataSource = DataSource()
self.addCleanup(self.__ds.engine.dispose)
+3 -2
View File
@@ -188,7 +188,8 @@ class TestRequestBuilding(RunLLMTestCase):
:return: The parsed request.
"""
stdout: io.StringIO = io.StringIO()
with redirect_stdout(stdout):
stderr: io.StringIO = io.StringIO()
with redirect_stdout(stdout), redirect_stderr(stderr):
run_llm.main([
str(self.__prompt), str(self.__input),
str(self.__archive_dir), "--dry-run"] + extra_argv)
@@ -239,7 +240,7 @@ class TestMainFlow(RunLLMTestCase):
str(self.__prompt), str(self.__input),
str(self.__archive_dir)]
self.__settings: config.Settings = config.Settings(
SQLALCHEMY_DATABASE_URL="sqlite://",
SQLALCHEMY_DATABASE_URI="sqlite://",
ANTHROPIC_API_KEY="test-key")
config.set_settings(self.__settings)
+1 -1
View File
@@ -50,7 +50,7 @@ class TestTallyAnnotations(unittest.TestCase):
run_dir.mkdir()
self.__runs.append(run_dir)
config.set_settings(config.Settings(
SQLALCHEMY_DATABASE_URL="sqlite://",
SQLALCHEMY_DATABASE_URI="sqlite://",
ANTHROPIC_API_KEY="test-key"))
self.__ds: DataSource = DataSource()
self.addCleanup(self.__ds.engine.dispose)
+1 -1
View File
@@ -39,7 +39,7 @@ class TestTallyCodings(unittest.TestCase):
self.__output_csv: Path \
= self.__dir / "results" / "codings.csv"
config.set_settings(config.Settings(
SQLALCHEMY_DATABASE_URL="sqlite://",
SQLALCHEMY_DATABASE_URI="sqlite://",
ANTHROPIC_API_KEY="test-key"))
self.__ds: DataSource = DataSource()
self.addCleanup(self.__ds.engine.dispose)