基礎から学ぶPython入門 90日コース | 実践開発 － Day 87：追加機能

Python 90日で身につけるPython
スポンサーリンク
スポンサーリンク

Day 87：追加機能 ― 「便利さと安心感を、少しずつコードに足していく」

Day 87では、最終プロジェクトに “便利さ” と “安心感” を足していきます。

今日のキーワードはこの5つ。

  • 検索
  • 集計
  • フィルター
  • エラー処理
  • ログ

昨日までで「入力 → 処理 → 保存」の一本の線は通りました。 今日はその上に、

「欲しいデータを探せる」 「数字をまとめて眺められる」 「条件で絞り込める」 「壊れたときに原因が分かる」 「あとから振り返れる」

そんな機能を、少しずつ重ねていきます。

検索機能 ― 「欲しいデータをピンポイントで取り出す」

何を検索したいかを決める

売上管理・分析システムを例にすると、 よくある検索の軸はこんな感じです。

  • 日付で検索（特定の日の売上）
  • 商品名で検索（特定の商品だけ）
  • 顧客で検索（特定の顧客の売上）

ここでは、まず 商品名で検索 する機能を作ってみます。

リポジトリに検索関数を追加する

# repositories/sales_repository.py
import sqlite3
from typing import List, Dict, Any

DB_PATH = "sales.db"

def find_sales_by_item(item: str) -> List[Dict[str, Any]]:
    """
    商品名で売上を検索します。

    item: 検索したい商品名
    戻り値: 該当する売上レコードのリスト（辞書形式）
    """
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row  # 行を辞書のように扱えるようにする
    cur = conn.cursor()

    cur.execute(
        "SELECT id, date, item, amount, customer_id FROM sales WHERE item = ?",
        (item,),
    )

    rows = cur.fetchall()
    conn.close()

    # sqlite3.Row を通常の dict に変換
    return [dict(row) for row in rows]
Python

サービス層から呼び出す

# services/search_service.py
from typing import List, Dict, Any
from repositories.sales_repository import find_sales_by_item

def search_sales_by_item(item: str) -> List[Dict[str, Any]]:
    """
    商品名で売上を検索するサービス関数。
    将来的にバリデーションやログなどを追加しやすいよう、
    リポジトリのラッパーとして用意しておきます。
    """
    if not item.strip():
        raise ValueError("商品名が空です")

    return find_sales_by_item(item)
Python

ポイント：

  • 検索ロジック（SQL）はリポジトリに置く
  • バリデーションや追加処理はサービス層で行う
  • 「どこで何をするか」を分けておくと、後から育てやすい

集計機能 ― 「数字をまとめて“意味のある情報”にする」

まずはシンプルな合計から

集計の第一歩は、やっぱり 合計 です。

# services/aggregation_service.py
import sqlite3
from typing import Dict, Any

DB_PATH = "sales.db"

def get_total_sales() -> int:
    """
    売上金額の合計を返します。
    """
    conn = sqlite3.connect(DB_PATH)
    cur = conn.cursor()

    cur.execute("SELECT SUM(amount) FROM sales")
    row = cur.fetchone()
    conn.close()

    # データがない場合は None になるので、0を返す
    return row[0] if row[0] is not None else 0
Python

もう一歩：日付ごとの集計

def get_daily_sales() -> Dict[str, int]:
    """
    日付ごとの売上合計を辞書で返します。

    戻り値の例:
    {
        "2024-01-01": 10000,
        "2024-01-02": 8000,
    }
    """
    conn = sqlite3.connect(DB_PATH)
    cur = conn.cursor()

    cur.execute(
        "SELECT date, SUM(amount) AS total FROM sales GROUP BY date ORDER BY date"
    )
    rows = cur.fetchall()
    conn.close()

    result: Dict[str, int] = {}
    for date, total in rows:
        result[date] = total

    return result
Python

ポイント：

  • 集計は「SQLでやるか」「Pythonでやるか」を選べる
  • 少ないデータならPythonでもOK、大量データならSQL側で集計したほうが効率的
  • まずはシンプルな合計から始めて、徐々に複雑な集計に広げていく

フィルター機能 ― 「条件で絞り込んで、見たいものだけを見る」

期間で絞り込むフィルター

「特定の期間だけの売上を見たい」というのは、 よくあるニーズです。

# services/filter_service.py
import sqlite3
from typing import List, Dict, Any

DB_PATH = "sales.db"

def filter_sales_by_date_range(start_date: str, end_date: str) -> List[Dict[str, Any]]:
    """
    指定した期間内の売上を取得します。

    start_date, end_date は "YYYY-MM-DD" 形式の文字列を想定しています。
    """
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    cur = conn.cursor()

    cur.execute(
        """
        SELECT id, date, item, amount, customer_id
        FROM sales
        WHERE date BETWEEN ? AND ?
        ORDER BY date
        """,
        (start_date, end_date),
    )

    rows = cur.fetchall()
    conn.close()

    return [dict(row) for row in rows]
Python

金額で絞り込むフィルター

def filter_sales_by_min_amount(min_amount: int) -> List[Dict[str, Any]]:
    """
    指定した金額以上の売上だけを取得します。
    """
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    cur = conn.cursor()

    cur.execute(
        """
        SELECT id, date, item, amount, customer_id
        FROM sales
        WHERE amount >= ?
        ORDER BY amount DESC
        """,
        (min_amount,),
    )

    rows = cur.fetchall()
    conn.close()

    return [dict(row) for row in rows]
Python

ポイント：

  • フィルターは「検索の一種」だが、条件が複数になることが多い
  • 条件を増やしすぎると複雑になるので、最初はシンプルなものから始める
  • フィルター機能は、レポートやダッシュボードの基礎になる

エラー処理 ― 「壊れたときに、静かに教えてくれる仕組み」

try/exceptで“落ちないコード”にする

これまでのコードは、 意図的に「例外をそのまま投げる」形にしていました。

Day 87では、 それを 「ユーザーに分かりやすく伝える形」 に変えていきます。

# app/cli_main.py
from services.csv_input import read_csv
from services.sales_processing import validate_and_convert
from repositories.sales_repository import save_sales

def run_import(path: str) -> None:
    """
    CSVインポート処理を実行します。
    エラーが起きても、メッセージを表示して終了します。
    """
    try:
        rows = read_csv(path)
        sales = validate_and_convert(rows)
        save_sales(sales)
        print("インポートが完了しました。")
    except FileNotFoundError:
        print(f"ファイルが見つかりません: {path}")
    except ValueError as e:
        print(f"データの形式に問題があります: {e}")
    except Exception as e:
        # 想定外のエラー
        print(f"予期しないエラーが発生しました: {e}")
Python

ポイント：

  • 「起こりうるエラー」を想像して、個別にメッセージを用意する
  • 最後に Exception をキャッチして、想定外のエラーも落ちないようにする
  • エラー処理は「ユーザーへのやさしさ」でもある

ログ ― 「あとから振り返れる“足跡”を残す」

loggingモジュールを使う

Python標準の logging モジュールを使うと、 簡単にログを残せます。

# config/logging_config.py
import logging

def setup_logging() -> None:
    """
    ログ設定を初期化します。
    """
    logging.basicConfig(
        level=logging.INFO,                # INFO以上のログを出す
        format="%(asctime)s [%(levelname)s] %(message)s",
        filename="app.log",               # ログをファイルに出力
    )
Python

処理の中でログを残す

# services/sales_processing.py
import logging
from typing import Any
from models.sale import Sale

logger = logging.getLogger(__name__)

def validate_and_convert(rows: list[dict[str, Any]]) -> list[Sale]:
    """
    CSVから読み込んだ行データをチェックし、
    Saleモデルのリストに変換します。
    """
    sales: list[Sale] = []
    logger.info("売上データのバリデーションを開始します。行数: %d", len(rows))

    for i, row in enumerate(rows, start=1):
        try:
            if not row.get("date"):
                raise ValueError("date が空です")
            if not row.get("item"):
                raise ValueError("item が空です")
            if not row.get("amount"):
                raise ValueError("amount が空です")

            amount_int = int(row["amount"])

            sale = Sale(
                id=None,
                date=row["date"],
                item=row["item"],
                amount=amount_int,
                customer_id=None,
            )
            sales.append(sale)
        except Exception as e:
            logger.error("行 %d でエラー: %s", i, e)
            raise

    logger.info("バリデーションが完了しました。有効な行数: %d", len(sales))
    return sales
Python

ポイント：

  • ログは「開発者・運用者のためのメッセージ」
  • エラーだけでなく、「処理が始まった」「処理が終わった」も記録しておくと、後から追いやすい
  • ログファイルは、トラブルシューティングのときの“タイムマシン”になる

Day 87ミニテンプレート ― 追加機能ひとまとめ

検索

# services/search_service.py
from repositories.sales_repository import find_sales_by_item

def search_sales_by_item(item: str):
    if not item.strip():
        raise ValueError("商品名が空です")
    return find_sales_by_item(item)
Python

集計

# services/aggregation_service.py
def get_total_sales() -> int:
    ...
def get_daily_sales() -> dict[str, int]:
    ...
Python

フィルター

# services/filter_service.py
def filter_sales_by_date_range(start_date: str, end_date: str):
    ...
def filter_sales_by_min_amount(min_amount: int):
    ...
Python

エラー処理＋ログの入口

# main.py
from config.logging_config import setup_logging
from app.cli_main import run_import

def main() -> None:
    setup_logging()
    run_import("sales_sample.csv")

if __name__ == "__main__":
    main()
Python

Day 87のまとめ ― 「便利さと安心感を、コードに少しずつ宿していく」

今日のテーマは、

  • 検索
  • 集計
  • フィルター
  • エラー処理
  • ログ

でした。

これらを足していくことで、 最終プロジェクトは、

「ただ動くだけのシステム」から、 「使っていて気持ちよく、壊れても原因が分かるシステム」

へと、静かに、でも確実に進化していきます。

Day 87は、 “便利さ” と “安心感” をコードに宿す日。

小さな機能の積み重ねが、 最終的には「自分でも誇れる作品」につながっていきます。

スポンサーリンク
タイトルとURLをコピーしました