公開APIの設計・実装に関するまとめ 〜APIキー管理・認証〜

from datetime import datetime
from enum import Enum
from sqlalchemy import Boolean, DateTime, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column
from shared.database import Base

class SCOPE_VALUE(str, Enum):
    READ_ONLY = "read_only"  # 読み専用
    READ_WRITE = "read_write"  # 読み書き可能

    @classmethod
    def to_list(cls) -> list[str]:
        return [s.value for s in cls]


class ApiKey(Base):
    __tablename__ = "api_keys"

    id: Mapped[int] = mapped_column(
        Integer, primary_key=True, autoincrement=True)  # APIキーの一意なID
    name: Mapped[str] = mapped_column(String(100), nullable=False)  # APIキー名称
    key_hash: Mapped[str] = mapped_column(
        String(64), nullable=False, unique=True, index=True)  # ハッシュ化したAPIキー文字列
    key_prefix: Mapped[str] = mapped_column(
        String(10), nullable=False, index=True)  # APIキーの先頭数文字
    is_active: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True)  # APIキーの有効フラグ
    scope: Mapped[str] = mapped_column(
        String(64), nullable=False, server_default=SCOPE_VALUE.READ_ONLY)  # 認可スコープ
    expires_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=True
    )  # 失効日時
    last_used_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=True
    )  # 最終使用日時
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=True, server_default=func.now()
    )  # 作成日時
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=True, server_default=func.now()
    )  # 更新日時

import os
from collections.abc import Generator

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker


DATABASE_URL = os.getenv(“DATABASE_URL”)

if not DATABASE_URL:
    raise RuntimeError("DATABASE_URLが設定されていません")


engine = create_engine(
    DATABASE_URL,
    pool_pre_ping=True,
)

SessionLocal = sessionmaker(
    bind=engine,
    autoflush=False,
    autocommit=False,
)


class Base(DeclarativeBase):
    pass


def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()

    try:
        yield db
    finally:
        db.close()

from fastapi import FastAPI, Request
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from api_key_manage.src.routes.api_keys import router as api_keys_router
from starlette import status
from starlette.responses import JSONResponse
from api_key_manage.src.services.api_keys import ApiKeyNotFoundError

app = FastAPI()
app.include_router(api_keys_router)


@app.exception_handler(RequestValidationError)
async def request_validation_exception_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
    """ リクエスト形式のバリデーションエラーを補足する共通ハンドラ """
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
        content={"detail": jsonable_encoder(exc.errors())},
    )


@app.exception_handler(ApiKeyNotFoundError)
async def api_key_not_found_handler(request: Request, exc: ApiKeyNotFoundError) -> JSONResponse:
    """ APIキーが見つからないエラーを補足する共通ハンドラ """
    return JSONResponse(
        status_code=status.HTTP_404_NOT_FOUND,
        content={"detail": "API key not found."},
    )


@app.exception_handler(Exception)
async def unhandled_exception_handler(request: Request, exc: Exception) -> JSONResponse:
    """ 想定外の例外を捕捉する共通ハンドラ """
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={"detail": "Internal Server Error."},
    )

from pydantic import BaseModel, ConfigDict, Field, model_validator
from typing import Self
from datetime import datetime, timezone
from api_key_manage.src.crud.api_keys import GetApiKeysCrudResult, CreateApiKeyCrudResult
from shared.models import ApiKey, SCOPE_VALUE


class ApiKeyOut(BaseModel):
    """ 公開してよいAPIキー情報 """
    model_config = ConfigDict(from_attributes=True)

    id: int
    name: str
    key_prefix: str
    is_active: bool
    scope: str
    last_used_at: datetime | None
    created_at: datetime
    updated_at: datetime
    # key_hash は入れない（内部専用なので外に出さない）


class GetApiKeysResponse(BaseModel):
    """ APIキー一覧取得レスポンス """
    api_keys: list[ApiKeyOut]

    @classmethod
    def from_crud(cls, result: GetApiKeysCrudResult) -> "GetApiKeysResponse":
        return cls(api_keys=[ApiKeyOut.model_validate(k) for k in result.api_keys])


class CreateApiKeyRequest(BaseModel):
    """ APIキー作成リクエスト """
    name: str = Field(min_length=1, max_length=100, description="APIキー名称")
    scope: SCOPE_VALUE = Field(
        min_length=1, max_length=100, description="認可スコープ")
    expires_at: datetime | None = Field(default=None, description="失効日時")

    @model_validator(mode="after")
    def check_expires_at(self) -> Self:
        if self.expires_at is None:
            return self
        if self.expires_at <= datetime.now(timezone.utc):
            raise ValueError("Invalid expires_at is specified.")
        return self


class CreateApiKeyResponse(BaseModel):
    """ APIキー作成レスポンス """
    id: int
    name: str
    api_key: str  # 生のAPIキー
    scope: str
    created_at: datetime

    @classmethod
    def from_crud(cls, result: CreateApiKeyCrudResult) -> "CreateApiKeyResponse":
        api_key = result.api_key
        original_key = result.original_key
        return cls(
            id=api_key.id,
            name=api_key.name,
            api_key=original_key,
            scope=api_key.scope,
            created_at=api_key.created_at,
        )


class UpdateApiKeyRequest(BaseModel):
    """ APIキー更新リクエスト """
    name: str = Field(min_length=1, max_length=255, description="APIキー名称")

    @model_validator(mode="after")
    def require_at_least_one_field(self) -> Self:
        if not self.model_fields_set:
            raise ValueError("At least one field must be provided.")
        return self


class UpdateApiKeyResponse(BaseModel):
    """ APIキー更新レスポンス """
    id: int
    updated_at: datetime

    @classmethod
    def from_model(cls, api_key: ApiKey) -> "UpdateApiKeyResponse":
        return cls(
            id=api_key.id,
            updated_at=api_key.updated_at
        )


class RevokeApiKeyResponse(BaseModel):
    """ APIキー無効化レスポンス """
    id: int
    updated_at: datetime

    @classmethod
    def from_model(cls, api_key: ApiKey) -> "RevokeApiKeyResponse":
        return cls(
            id=api_key.id,
            updated_at=api_key.updated_at
        )

from fastapi import APIRouter, Depends
from api_key_manage.src.schemas import api_keys as schemas
from api_key_manage.src.services.api_keys import get_api_keys_service, create_api_key_service, update_api_key_service, revoke_api_key_service
from shared.database import get_db
from sqlalchemy.orm import Session

router = APIRouter(prefix="/api_keys", tags=["api_keys"])


@router.get("", response_model=schemas.GetApiKeysResponse)
async def get_api_keys(active_filter: bool = False, db: Session = Depends(get_db)) -> schemas.GetApiKeysResponse:
    """ APIキー一覧取得エンドポイント """
    result = get_api_keys_service(db=db, active_filter=active_filter)

    return schemas.GetApiKeysResponse.from_crud(result)


@router.post("", response_model=schemas.CreateApiKeyResponse)
async def create_api_key(request: schemas.CreateApiKeyRequest, db: Session = Depends(get_db)) -> schemas.CreateApiKeyResponse:
    """ APIキー作成エンドポイント """
    name = request.name.strip()
    scope = request.scope.strip()
    expires_at = request.expires_at
    result = create_api_key_service(
        name=name, scope=scope, expires_at=expires_at, db=db)

    return schemas.CreateApiKeyResponse.from_crud(result)


@router.patch("/{key_id}", response_model=schemas.UpdateApiKeyResponse)
async def update_api_key(
        key_id: int,
        request: schemas.UpdateApiKeyRequest,
        db: Session = Depends(get_db)) -> schemas.UpdateApiKeyResponse:
    """ APIキー更新エンドポイント """
    name = request.name.strip()
    api_key = update_api_key_service(key_id=key_id, name=name, db=db)

    return schemas.UpdateApiKeyResponse.from_model(api_key)


@router.post("/{key_id}/revoke", response_model=schemas.RevokeApiKeyResponse)
async def revoke_api_key(key_id: int, db: Session = Depends(get_db)) -> schemas.RevokeApiKeyResponse:
    """ APIキー無効化エンドポイント """
    api_key = revoke_api_key_service(key_id=key_id, db=db)

    return schemas.RevokeApiKeyResponse.from_model(api_key)

from datetime import datetime, timezone
from sqlalchemy.orm import Session
from api_key_manage.src.crud.api_keys import GetApiKeysCrudResult, CreateApiKeyCrudResult, get_api_key_from_id_crud, get_api_keys_crud, create_api_key_crud
from shared.models import ApiKey


class ApiKeyNotFoundError(Exception):
    """ 指定のAPIキーが見つからないエラー """


def _get_now_utc() -> datetime:
    """ 現在時刻（UTC）を取得する """
    return datetime.now(timezone.utc)


def get_api_keys_service(db: Session, active_filter: bool = False) -> GetApiKeysCrudResult:
    """ APIキー一覧取得処理 """
    result = get_api_keys_crud(db, active_filter)

    return result


def create_api_key_service(name: str, scope: str, expires_at: datetime | None, db: Session) -> CreateApiKeyCrudResult:
    """ APIキー作成処理 """
    try:
        result = create_api_key_crud(name, scope, expires_at, db)
        db.commit()
        return result
    except Exception:
        db.rollback()
        raise


def update_api_key_service(key_id: int, name: str, db: Session) -> ApiKey:
    """ APIキー更新処理 """
    try:
        # 指定のAPIキーを取得
        api_key = get_api_key_from_id_crud(key_id, db)
        if not api_key:
            raise ApiKeyNotFoundError()

        # APIキー情報を更新する
        api_key.name = name
        api_key.updated_at = _get_now_utc()
        db.commit()

        return api_key
    except Exception:
        db.rollback()
        raise


def revoke_api_key_service(key_id: int, db: Session) -> ApiKey:
    """ APIキー無効化処理 """
    try:
        # 指定のAPIキーを取得
        api_key = get_api_key_from_id_crud(key_id, db)
        if not api_key:
            raise ApiKeyNotFoundError()

        if api_key.is_active:
            # キーが有効な場合だけ更新日時を最新化
            api_key.updated_at = _get_now_utc()
        api_key.is_active = False
        db.commit()

        return api_key
    except Exception:
        db.rollback()
        raise

import hashlib
import secrets
from dataclasses import dataclass
from datetime import datetime
from shared.models import ApiKey
from sqlalchemy import select
from sqlalchemy.orm import Session

API_KEY_PREFIX = "sk-"
API_KEY_SHOW_LENGTH = 8


# crud用クラスの定義
# crud層とservice層の間だけでやり取りする内部の戻り値なので外部入力の検証（pydantic）は不要。
# むしろ、ApiKeyクラスはSQLAlchemyのORMモデルなので、pydanticのBaseModelとして定義するとunable to generate pydantic-core schema for <class 'shared.models.ApiKey'>で失敗


@dataclass
class GetApiKeysCrudResult:
    """ APIキー一覧取得CRUDの返却値用クラス """
    api_keys: list[ApiKey]


@dataclass
class CreateApiKeyCrudResult:
    """ APIキー作成CRUDの返却値用クラス """
    api_key: ApiKey
    original_key: str


def get_api_key_from_id_crud(key_id: int, db: Session) -> ApiKey:
    """ IDからAPIキーを1件取得するDB処理 """
    return db.scalar(select(ApiKey).where(ApiKey.id == key_id))


def get_api_keys_crud(db: Session, active_filter: bool = False) -> GetApiKeysCrudResult:
    """ 現在作成されているAPIキー一覧を取得するDB処理 """
    stmt = select(ApiKey)
    if active_filter:
        # 有効なAPIキーのみ取得するwhere句を追加
        stmt = stmt.where(ApiKey.is_active == True)
    return GetApiKeysCrudResult(db.scalars(stmt).all())


def create_api_key_crud(name: str, scope: str, expires_at: datetime | None, db: Session) -> CreateApiKeyCrudResult:
    """ APIキー作成のDB処理 """
    token = secrets.token_urlsafe(32)
    # 生のAPIキー
    original_key = API_KEY_PREFIX + token
    # SHA256でAPIキーをハッシュ変換
    key_hash = hashlib.sha256(original_key.encode("utf-8")).hexdigest()

    api_key = ApiKey(
        name=name,
        key_hash=key_hash,
        key_prefix=original_key[:API_KEY_SHOW_LENGTH],
        is_active=True,
        scope=scope,
        expires_at=expires_at,
        last_used_at=None,
    )
    db.add(api_key)
    return CreateApiKeyCrudResult(api_key=api_key, original_key=original_key)

from fastapi import FastAPI, Request
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from public_api.src.services.scores import ScoresForSubjectNotFoundError, DuplicateStudentScoreError
from public_api.src.routes.scores import router as scores_router
from starlette import status
from starlette.responses import JSONResponse

app = FastAPI()
app.include_router(scores_router)


@app.exception_handler(RequestValidationError)
async def request_validation_exception_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
    """ リクエスト形式のバリデーションエラーを補足する共通ハンドラ """
    logger.warning(
        "Request validation error: method=%s path=%s errors=%s",
        request.method,
        request.url.path,
        exc.errors(),
        exc_info=exc,
    )
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
        content={"detail": jsonable_encoder(exc.errors())},
    )


@app.exception_handler(ScoresForSubjectNotFoundError)
async def score_for_subject_not_found_handler(request: Request, exc: ScoresForSubjectNotFoundError) -> JSONResponse:
    """ スコアレコードが見つからないエラーを補足する共通ハンドラ """
    logger.warning(
        "Scores for the subject not found: method=%s path=%s",
        request.method,
        request.url.path,
        exc_info=exc,
    )
    return JSONResponse(
        status_code=status.HTTP_404_NOT_FOUND,
        content={"detail": "Scores for the subject not found."},
    )


@app.exception_handler(DuplicateStudentScoreError)
async def duplicate_student_score_handler(request: Request, exc: DuplicateStudentScoreError) -> JSONResponse:
    """ スコアレコードの生徒IDが重複しているエラーを補足する共通ハンドラ """
    logger.warning(
        "This student's score record already exists: method=%s path=%s",
        request.method,
        request.url.path,
        exc_info=exc,
    )
    return JSONResponse(
        status_code=status.HTTP_409_CONFLICT,
        content={"detail": "This student's score record already exists."},
    )


@app.exception_handler(Exception)
async def unhandled_exception_handler(request: Request, exc: Exception) -> JSONResponse:
    """ 想定外の例外を捕捉する共通ハンドラ """
    logger.error(
        "Unhandled exception: method=%s path=%s",
        request.method,
        request.url.path,
        exc_info=exc,
    )

    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={"detail": "Internal Server Error."},
    )

import hashlib
from datetime import datetime, timezone
from fastapi import Depends, Header, HTTPException
from shared.database import get_db
from shared.models import ApiKey, SCOPE_VALUE
from sqlalchemy.orm import Session
from starlette import status
from public_api.src.crud.auth import get_api_key_by_key_hash_crud
from typing import Callable

API_KEY_PREFIX = "sk-"


def _get_now_utc() -> datetime:
    """ 現在時刻（UTC）を取得する """
    return datetime.now(timezone.utc)


def _authenticate_api_key(
    api_key: str | None = Header(default=None, alias="X-API-Key"),
    db: Session = Depends(get_db),
) -> ApiKey:
    """ APIキー文字列の検証を行い正当なAPIキーレコードを返却する """
    try:
        # APIキーが付加されていない
        if not api_key:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="X-Api-Key header is required.",
            )

        # APIキーの形式不正
        if not api_key.startswith(API_KEY_PREFIX):
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid API key.",
            )

        # APIキーをハッシュ化し、それを元にAPIキーレコードを取得する
        key_hash = hashlib.sha256(api_key.encode("utf-8")).hexdigest()
        api_key_recode = get_api_key_by_key_hash_crud(key_hash, db)

        # APIキー文字列と一致するAPIキーレコードが存在しない
        if not api_key_recode:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid API key.",
            )

        # APIキーが無効
        if not api_key_recode.is_active:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="API key has been revoked.",
            )

        # APIキーが有効期限切れ
        if api_key_recode.expires_at and api_key_recode.expires_at <= _get_now_utc():
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="API key has been expired.",
            )

        # チェックOKなのでAPIキーのlast_used_atを現在日時に更新
        api_key_recode.last_used_at = _get_now_utc()
        db.commit()
        return api_key_recode
    except Exception:
        db.rollback()
        raise


def authorize_api_key(*allowed_scopes: SCOPE_VALUE) -> Callable[..., ApiKey]:
    """ 認証 + スコープに基づく認可 を行う """
    # APIキーレコードは依存関係として取得
    def _checker(api_key_recode: ApiKey = Depends(_authenticate_api_key)) -> ApiKey:
        # APIキーレコードのscopeがallowed_scopesリストに含まれていなければ認可失敗
        if api_key_recode.scope not in allowed_scopes:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"This operation requires one of the following scopes: {','.join(allowed_scopes)}."
            )
        return api_key_recode

    return _checker

from shared.models import ApiKey
from sqlalchemy import select
from sqlalchemy.orm import Session


def get_api_key_by_key_hash_crud(key_hash: str, db: Session) -> ApiKey | None:
    """ APIキーハッシュ文字列からAPIキーレコードを取得するDB処理 """
    return db.scalar(select(ApiKey).where(ApiKey.key_hash == key_hash))

from fastapi import APIRouter, Depends
from typing import Literal
from public_api.src.schemas import scores as schemas
from public_api.src.services.auth import authorize_api_key
from public_api.src.services.scores import get_one_subject_scores, calc_statistics, create_score_service
from sqlalchemy.orm import Session
from shared.database import get_db
from shared.models import ApiKey, SCOPE_VALUE

router = APIRouter(prefix="/scores", tags=["scores"])


@router.get("/statistics", response_model=schemas.ScoreStatisticsResponse)
async def get_one_subject_statistics(
    subject: Literal["japanese", "math", "english",
                     "physics", "chemistry", "world_history"],
    _: ApiKey = Depends(authorize_api_key(
        SCOPE_VALUE.READ_ONLY, SCOPE_VALUE.READ_WRITE)),
    db: Session = Depends(get_db)
) -> schemas.ScoreStatisticsResponse:
    """ 1教科のスコアの統計量の計算結果を取得する """
    scores = get_one_subject_scores(subject, db)
    result = calc_statistics(scores.scores)

    return schemas.ScoreStatisticsResponse.from_result(subject, result)


@router.post("", response_model=schemas.CreateScoreResponse)
async def create_score(
    request: schemas.CreateScoreRequest,
    _: ApiKey = Depends(authorize_api_key(SCOPE_VALUE.READ_WRITE)),
    db: Session = Depends(get_db),
) -> schemas.CreateScoreResponse:
    """ スコアレコードを1件作成する """
    result = create_score_service(
        student_id=request.student_id,
        japanese=request.japanese,
        math=request.math,
        english=request.english,
        physics=request.physics,
        chemistry=request.chemistry,
        world_history=request.world_history,
        db=db,
    )

    return schemas.CreateScoreResponse.from_crud(result)

from pydantic import BaseModel, Field
from public_api.src.services.scores import ScoreStatisticsResult
from public_api.src.crud.scores import CreateScoreCrudResult


class StudentDeviationValueResponse(BaseModel):
    """ 生徒1人分の偏差値（レスポンス用） """
    student_id: int
    deviation_value: float


class ScoreStatisticsResponse(BaseModel):
    """ スコア統計量取得レスポンス """
    subject: str
    count: int
    max: int
    min: int
    average: float
    deviation_values: list[StudentDeviationValueResponse]

    @classmethod
    def from_result(cls, subject: str, result: ScoreStatisticsResult) -> "ScoreStatisticsResponse":
        return cls(
            subject=subject,
            count=result.count,
            max=result.max,
            min=result.min,
            average=result.average,
            deviation_values=[
                StudentDeviationValueResponse(student_id=d.student_id, deviation_value=d.deviation_value) for d in result.deviation_values
            ]
        )


class CreateScoreRequest(BaseModel):
    """ スコアレコード作成リクエスト （各スコア100点満点）"""
    student_id: int = Field(ge=0, description="生徒ID")
    japanese: int = Field(ge=0, le=100, description="国語のスコア")
    math: int = Field(ge=0, le=100, description="数学のスコア")
    english: int = Field(ge=0, le=100, description="英語のスコア")
    physics: int = Field(ge=0, le=100, description="物理のスコア")
    chemistry: int = Field(ge=0, le=100, description="化学のスコア")
    world_history: int = Field(ge=0, le=100, description="世界史のスコア")


class CreateScoreResponse(BaseModel):
    """ スコアレコード作成レスポンス """
    id: int
    student_id: int
    japanese: int
    math: int
    english: int
    physics: int
    chemistry: int
    world_history: int

    @classmethod
    def from_crud(cls, result: CreateScoreCrudResult) -> "CreateScoreResponse":
        score = result.score
        return cls(
            id=score.id,
            student_id=score.student_id,
            japanese=score.japanese,
            math=score.math,
            english=score.english,
            physics=score.physics,
            chemistry=score.chemistry,
            world_history=score.world_history,
        )

import statistics
from dataclasses import dataclass
from sqlalchemy.orm import Session
from public_api.src.crud.scores import StudentScore, GetScoresCrudResult, CreateScoreCrudResult, get_scores_by_subject, get_score_by_student_id, create_score_crud


class ScoresForSubjectNotFoundError(Exception):
    """ 指定教科に対するスコアレコードが見つからないエラー """


class DuplicateStudentScoreError(Exception):
    """ スコアレコードの生徒IDが重複しているエラー """


@dataclass
class StudentDeviationValue:
    """ 生徒1人分の偏差値 """
    student_id: int
    deviation_value: float


@dataclass
class ScoreStatisticsResult:
    """ 統計量計算処理の返却値用クラス """
    count: int
    max: int
    min: int
    average: float
    deviation_values: list[StudentDeviationValue]


def get_one_subject_scores(subject: str, db: Session) -> GetScoresCrudResult:
    """ 1教科の全スコアレコードを取得する処理 """
    scores = get_scores_by_subject(subject, db)
    if not scores:
        raise ScoresForSubjectNotFoundError

    return scores


def calc_statistics(scores: list[StudentScore]) -> dict:
    """
        指定教科の統計量を計算して返却する
        スコアの全体統計（最大値、最小値、平均値）と生徒毎の偏差値を計算する
    """
    # 点数のリスト
    values = [s.score for s in scores]
    average = round(statistics.mean(values), 1)
    standard_deviation = statistics.pstdev(values)

    def _deviation_value(score: int) -> float:
        # 偏差値の計算
        if standard_deviation == 0:
            return 50.0
        else:
            return round(50 + 10 * (score - average) / standard_deviation, 1)

    deviation_values = [StudentDeviationValue(
        student_id=s.student_id, deviation_value=_deviation_value(s.score)) for s in scores]

    return ScoreStatisticsResult(
        count=len(values),
        max=max(values),
        min=min(values),
        average=average,
        deviation_values=deviation_values,
    )


def create_score_service(
    student_id: int,
    japanese: int,
    math: int,
    english: int,
    physics: int,
    chemistry: int,
    world_history: int,
    db: Session,
) -> CreateScoreCrudResult:
    """ スコアレコードを1件作成する処理 """
    try:
        exist_score = get_score_by_student_id(student_id, db)
        if exist_score:
            raise DuplicateStudentScoreError
        result = create_score_crud(
            student_id=student_id,
            japanese=japanese,
            math=math,
            english=english,
            physics=physics,
            chemistry=chemistry,
            world_history=world_history,
            db=db,
        )
        db.commit()
        return result
    except Exception:
        db.rollback()
        raise

from dataclasses import dataclass
from sqlalchemy import select, Sequence, RowMapping
from sqlalchemy.orm import Session
from shared.models import Score

_SCORE_COLUMNS = {
    "student_id": Score.student_id,
    "japanese": Score.japanese,
    "math": Score.math,
    "english": Score.english,
    "physics": Score.physics,
    "chemistry": Score.chemistry,
    "world_history": Score.world_history,
}


@dataclass
class StudentScore:
    """ 生徒IDとスコアのセットデータ """
    student_id: int
    score: int

    @classmethod
    def from_query_response(cls, subject: str, row: Sequence[RowMapping]) -> "StudentScore":
        return cls(
            student_id=row["student_id"],
            score=row[subject],
        )


@dataclass
class GetScoresCrudResult:
    """ 指定教科の全スコアレコードを取得するCRUDの返却値用クラス """
    scores: list[StudentScore]


@dataclass
class CreateScoreCrudResult:
    """ スコアレコード作成CRUDの返却値用クラス """
    score: Score


def get_score_by_student_id(student_id: int, db: Session) -> Score:
    """ 生徒IDからスコアレコードを取得するDB処理 """
    return db.scalar(select(Score).where(Score.student_id == student_id))


def get_scores_by_subject(subject: str, db: Session) -> GetScoresCrudResult:
    """
        指定教科の全スコアレコードを取得するDB処理
        (student_id, score)のセットとして取得する
    """
    columns = [_SCORE_COLUMNS[k] for k in ["student_id", subject]]
    rows = db.execute(select(*columns)).mappings().all()

    return GetScoresCrudResult(
        scores=[StudentScore.from_query_response(subject, r) for r in rows])


def create_score_crud(
    student_id: int,
    japanese: int,
    math: int,
    english: int,
    physics: int,
    chemistry: int,
    world_history: int,
    db: Session,
) -> CreateScoreCrudResult:
    """ スコアレコード作成のDB処理 """
    score = Score(
        student_id=student_id,
        japanese=japanese,
        math=math,
        english=english,
        physics=physics,
        chemistry=chemistry,
        world_history=world_history,
    )
    db.add(score)
    return CreateScoreCrudResult(score=score)