119 lines
3.8 KiB
Python
119 lines
3.8 KiB
Python
from fastapi import APIRouter
|
||
from pydantic import BaseModel, field_validator
|
||
from enum import Enum
|
||
from psycopg.errors import UniqueViolation, CheckViolation, StringDataRightTruncation
|
||
from app.core.database import get_connection
|
||
from app.core.exceptions import BusinessException
|
||
import psycopg
|
||
|
||
|
||
router = APIRouter(prefix="/accounts", tags=["科目マスタ"])
|
||
|
||
|
||
# ======================================================
|
||
# Enum：科目種別
|
||
# ======================================================
|
||
class AccountType(str, Enum):
|
||
asset = "asset"
|
||
liability = "liability"
|
||
equity = "equity"
|
||
revenue = "revenue"
|
||
expense = "expense"
|
||
|
||
|
||
# ======================================================
|
||
# Model：登録入力用
|
||
# ======================================================
|
||
class AccountCreate(BaseModel):
|
||
account_code: str
|
||
account_name: str
|
||
account_type: AccountType
|
||
|
||
@field_validator("account_code")
|
||
@classmethod
|
||
def validate_code(cls, v: str):
|
||
if not v.isdigit():
|
||
raise BusinessException(
|
||
error_code="ACCOUNT_CODE_INVALID",
|
||
message="科目コードは数字のみで入力してください"
|
||
)
|
||
if len(v) > 20:
|
||
raise BusinessException(
|
||
error_code="ACCOUNT_CODE_TOO_LONG",
|
||
message="科目コードは20文字以内で入力してください"
|
||
)
|
||
return v
|
||
|
||
@field_validator("account_name")
|
||
@classmethod
|
||
def validate_name(cls, v: str):
|
||
if not v.strip():
|
||
raise BusinessException(
|
||
error_code="ACCOUNT_NAME_EMPTY",
|
||
message="科目名を入力してください"
|
||
)
|
||
if len(v) > 100:
|
||
raise BusinessException(
|
||
error_code="ACCOUNT_NAME_TOO_LONG",
|
||
message="科目名は100文字以内で入力してください"
|
||
)
|
||
return v
|
||
|
||
|
||
# ======================================================
|
||
# GET：科目一覧取得
|
||
# ======================================================
|
||
@router.get("/", summary="科目一覧取得")
|
||
def list_accounts():
|
||
with get_connection() as conn, conn.cursor() as cur:
|
||
cur.execute("""
|
||
SELECT account_id, account_code, account_name, account_type, is_active
|
||
FROM accounts
|
||
ORDER BY account_code
|
||
""")
|
||
rows = cur.fetchall()
|
||
|
||
if not rows:
|
||
raise BusinessException(
|
||
error_code="ACCOUNT_NOT_FOUND",
|
||
message="科目が登録されていません"
|
||
)
|
||
|
||
return rows
|
||
|
||
|
||
# ======================================================
|
||
# POST：科目登録
|
||
# ======================================================
|
||
@router.post("/", summary="科目登録", status_code=201)
|
||
def create_account(data: AccountCreate):
|
||
try:
|
||
with get_connection() as conn, conn.cursor() as cur:
|
||
cur.execute("""
|
||
INSERT INTO accounts (account_code, account_name, account_type)
|
||
VALUES (%s, %s, %s)
|
||
RETURNING account_id
|
||
""", (data.account_code, data.account_name, data.account_type.value))
|
||
|
||
new_id = cur.fetchone()["account_id"]
|
||
conn.commit()
|
||
return {"account_id": new_id}
|
||
|
||
except UniqueViolation:
|
||
raise BusinessException(
|
||
error_code="ACCOUNT_DUPLICATE",
|
||
message="この科目コードは既に登録されています"
|
||
)
|
||
|
||
except StringDataRightTruncation:
|
||
raise BusinessException(
|
||
error_code="ACCOUNT_VALUE_TOO_LONG",
|
||
message="入力文字数が制限を超えています"
|
||
)
|
||
|
||
except CheckViolation:
|
||
raise BusinessException(
|
||
error_code="ACCOUNT_VALUE_INVALID",
|
||
message="入力値が不正です"
|
||
)
|