fix: add parameter validation to admin API

Add Query parameter validation with regex patterns for date fields (YYYYMMDD format) and numeric constraints for limit parameter. Update JobLogResponse to use datetime types instead of strings for proper date serialization.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
zephyrdark 2026-02-03 00:00:18 +09:00
parent d2c5f91b2b
commit aec8541563

View File

@ -2,8 +2,9 @@
Admin API for data collection management.
"""
from typing import List
from datetime import datetime
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from pydantic import BaseModel
@ -24,8 +25,8 @@ class JobLogResponse(BaseModel):
id: int
job_name: str
status: str
started_at: str
finished_at: str | None
started_at: datetime
finished_at: datetime | None
records_count: int | None
error_msg: str | None
@ -42,7 +43,7 @@ class CollectResponse(BaseModel):
async def collect_stocks(
current_user: CurrentUser,
db: Session = Depends(get_db),
biz_day: str = None,
biz_day: str = Query(None, regex=r"^\d{8}$", description="Business day in YYYYMMDD format"),
):
"""Collect stock master data from KRX."""
try:
@ -60,7 +61,7 @@ async def collect_stocks(
async def collect_sectors(
current_user: CurrentUser,
db: Session = Depends(get_db),
biz_day: str = None,
biz_day: str = Query(None, regex=r"^\d{8}$", description="Business day in YYYYMMDD format"),
):
"""Collect sector classification data from WISEindex."""
try:
@ -78,8 +79,8 @@ async def collect_sectors(
async def collect_prices(
current_user: CurrentUser,
db: Session = Depends(get_db),
start_date: str = None,
end_date: str = None,
start_date: str = Query(None, regex=r"^\d{8}$", description="Start date in YYYYMMDD format"),
end_date: str = Query(None, regex=r"^\d{8}$", description="End date in YYYYMMDD format"),
):
"""Collect price data using pykrx."""
try:
@ -97,7 +98,7 @@ async def collect_prices(
async def collect_valuations(
current_user: CurrentUser,
db: Session = Depends(get_db),
biz_day: str = None,
biz_day: str = Query(None, regex=r"^\d{8}$", description="Business day in YYYYMMDD format"),
):
"""Collect valuation data from KRX."""
try:
@ -115,7 +116,7 @@ async def collect_valuations(
async def get_collection_status(
current_user: CurrentUser,
db: Session = Depends(get_db),
limit: int = 20,
limit: int = Query(20, gt=0, le=100, description="Number of records to return"),
):
"""Get recent job execution status."""
jobs = (