feat: DatabaseHealthCheck と app.py の DB_ADAPTER=sqlite 配線を追加する (#16)

.env.example

@@ -4,9 +4,26 @@ APP_DEBUG=false
 APP_NAME=nene2-python
 
 # データベース設定
+# sqlite: インメモリ（テスト用）
+# DB_ADAPTER=sqlite
+# DB_NAME=:memory:
+
+# sqlite: ファイル永続化
+# DB_ADAPTER=sqlite
+# DB_NAME=data/nene2.db
+
+# MySQL
+# DB_ADAPTER=mysql
+# DB_NAME=nene2
+# DB_HOST=localhost
+# DB_PORT=3306
+# DB_USER=nene2
+# DB_PASSWORD=secret
+
+# デフォルト値
 DB_ADAPTER=sqlite
 DB_NAME=:memory:
-DB_HOST=
-DB_PORT=0
+DB_HOST=localhost
+DB_PORT=3306
 DB_USER=
 DB_PASSWORD=

src/example/app.py

@@ -2,25 +2,35 @@
 
 from fastapi import FastAPI
 from fastapi.responses import JSONResponse
+from sqlalchemy import create_engine
+from sqlalchemy.pool import StaticPool
 
 from nene2.config import AppSettings
+from nene2.database import (
+    DatabaseHealthCheck,
+    DatabaseQueryExecutorInterface,
+    SqlAlchemyQueryExecutor,
+)
 from nene2.http import HealthStatus
 from nene2.middleware import ErrorHandlerMiddleware
 from nene2.validation.exceptions import ValidationException
 
 from .note.exceptions import NoteNotFoundExceptionHandler
 from .note.handler import make_note_router
-from .note.repository import InMemoryNoteRepository
+from .note.repository import InMemoryNoteRepository, NoteRepositoryInterface
+from .note.sqlite_repository import SqliteNoteRepository
 from .note.use_case import (
     CreateNoteUseCase,
     DeleteNoteUseCase,
     GetNoteUseCase,
     ListNotesUseCase,
     UpdateNoteUseCase,
 )
+from .schema import ensure_schema
 from .tag.exceptions import TagNotFoundExceptionHandler
 from .tag.handler import make_tag_router
-from .tag.repository import InMemoryTagRepository
+from .tag.repository import InMemoryTagRepository, TagRepositoryInterface
+from .tag.sqlite_repository import SqliteTagRepository
 from .tag.use_case import (
     CreateTagUseCase,
     DeleteTagUseCase,
@@ -30,6 +40,23 @@
 )
 
 
+def _build_repositories(
+    cfg: AppSettings,
+) -> tuple[NoteRepositoryInterface, TagRepositoryInterface, DatabaseQueryExecutorInterface | None]:
+    """Build repositories based on DB_ADAPTER setting."""
+    if cfg.db_adapter == "sqlite":
+        is_memory = cfg.db_name == ":memory:"
+        engine = create_engine(
+            cfg.db_url,
+            connect_args={"check_same_thread": False},
+            poolclass=StaticPool if is_memory else None,
+        )
+        ensure_schema(engine)
+        executor = SqlAlchemyQueryExecutor(engine)
+        return SqliteNoteRepository(executor), SqliteTagRepository(executor), executor
+    return InMemoryNoteRepository(), InMemoryTagRepository(), None
+
+
 def create_app(settings: AppSettings | None = None) -> FastAPI:
     cfg = settings or AppSettings()
 
@@ -50,8 +77,8 @@ def create_app(settings: AppSettings | None = None) -> FastAPI:
         ErrorHandlerMiddleware.handle_validation_exception,
     )
 
-    # Wire note domain
-    note_repo = InMemoryNoteRepository()
+    note_repo, tag_repo, db_executor = _build_repositories(cfg)
+
     app.include_router(
         make_note_router(
             ListNotesUseCase(note_repo),
@@ -62,8 +89,6 @@ def create_app(settings: AppSettings | None = None) -> FastAPI:
         )
     )
 
-    # Wire tag domain
-    tag_repo = InMemoryTagRepository()
     app.include_router(
         make_tag_router(
             ListTagsUseCase(tag_repo),
@@ -74,9 +99,11 @@ def create_app(settings: AppSettings | None = None) -> FastAPI:
         )
     )
 
+    db_health = DatabaseHealthCheck(db_executor) if db_executor else None
+
     @app.get("/health", tags=["system"], summary="Health check")
     async def health() -> JSONResponse:
-        status = HealthStatus(status="ok")
+        status = db_health.check() if db_health else HealthStatus(status="ok")
         code = 200 if status.is_healthy else 503
         return JSONResponse({"status": status.status, "checks": status.checks}, status_code=code)
 

src/example/schema.py

@@ -0,0 +1,28 @@
+"""Schema bootstrap — ensures tables exist for in-memory SQLite or fresh deployments.
+
+Production: use `alembic upgrade head` before starting the app.
+In-memory SQLite (test/dev): tables are created here with IF NOT EXISTS.
+"""
+
+from sqlalchemy import Engine, text
+
+
+def ensure_schema(engine: Engine) -> None:
+    """Create tables if they do not already exist (idempotent)."""
+    with engine.begin() as conn:
+        conn.execute(text(
+            "CREATE TABLE IF NOT EXISTS notes ("
+            "id INTEGER PRIMARY KEY AUTOINCREMENT,"
+            "title TEXT NOT NULL,"
+            "body TEXT NOT NULL,"
+            "created_at DATETIME DEFAULT CURRENT_TIMESTAMP,"
+            "updated_at DATETIME DEFAULT CURRENT_TIMESTAMP"
+            ")"
+        ))
+        conn.execute(text(
+            "CREATE TABLE IF NOT EXISTS tags ("
+            "id INTEGER PRIMARY KEY AUTOINCREMENT,"
+            "name TEXT NOT NULL UNIQUE,"
+            "created_at DATETIME DEFAULT CURRENT_TIMESTAMP"
+            ")"
+        ))

src/nene2/config/settings.py

@@ -1,6 +1,6 @@
 """Typed application settings loaded from environment variables."""
 
-from pydantic import field_validator
+from pydantic import SecretStr, field_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
@@ -15,10 +15,10 @@ class AppSettings(BaseSettings):
 
     db_adapter: str = "sqlite"
     db_name: str = ":memory:"
-    db_host: str = ""
-    db_port: int = 0
+    db_host: str = "localhost"
+    db_port: int = 3306
     db_user: str = ""
-    db_password: str = ""
+    db_password: SecretStr = SecretStr("")
 
     @field_validator("db_adapter")
     @classmethod
@@ -27,3 +27,14 @@ def validate_adapter(cls, v: str) -> str:
         if v not in allowed:
             raise ValueError(f"db_adapter must be one of {allowed}")
         return v
+
+    @property
+    def db_url(self) -> str:
+        """Build a SQLAlchemy connection URL from adapter + credentials."""
+        if self.db_adapter == "sqlite":
+            return f"sqlite:///{self.db_name}"
+        password = self.db_password.get_secret_value()
+        port = self.db_port
+        if self.db_adapter == "mysql":
+            return f"mysql+pymysql://{self.db_user}:{password}@{self.db_host}:{port}/{self.db_name}"
+        return f"postgresql+psycopg2://{self.db_user}:{password}@{self.db_host}:{port}/{self.db_name}"

src/nene2/database/__init__.py

@@ -1,11 +1,13 @@
 """NENE2 database abstraction layer."""
 
 from .exceptions import DatabaseConnectionException
+from .health import DatabaseHealthCheck
 from .interfaces import DatabaseQueryExecutorInterface, DatabaseTransactionManagerInterface
 from .sqlalchemy_executor import SqlAlchemyQueryExecutor, SqlAlchemyTransactionManager
 
 __all__ = [
     "DatabaseConnectionException",
+    "DatabaseHealthCheck",
     "DatabaseQueryExecutorInterface",
     "DatabaseTransactionManagerInterface",
     "SqlAlchemyQueryExecutor",

src/nene2/database/health.py

@@ -0,0 +1,19 @@
+"""Database health check — verifies DB connectivity for /health endpoint."""
+
+from nene2.http import HealthStatus
+
+from .interfaces import DatabaseQueryExecutorInterface
+
+
+class DatabaseHealthCheck:
+    """Check database connectivity by executing a lightweight query."""
+
+    def __init__(self, executor: DatabaseQueryExecutorInterface) -> None:
+        self._executor = executor
+
+    def check(self) -> HealthStatus:
+        try:
+            self._executor.fetch_one("SELECT 1 AS ok")
+            return HealthStatus(status="ok", checks={"database": "ok"})
+        except Exception:
+            return HealthStatus(status="degraded", checks={"database": "error"})