Python＋SQLiteで障害ログ検索APIを作ってみた｜CSV集計から一段進めるローカルツール

前回は、CSV形式の障害ログからERRORだけを抽出して集計する小さなCLIツールを作りました。そこから一段進めて、今回はCSVをSQLiteへ取り込み、HTTP経由で条件検索できるローカルAPIを作って動かしてみました。

今回の狙いは、単発の集計ではなく「ログを保存して、あとから別の条件で何度でも検索できる形」にすることです。障害調査では、一度集計して終わりではなく、対象システムや時間帯を変えながら何度も見直すことがあります。そうした場面では、CSVを毎回読み直すより、ローカルDBへ取り込んで検索できる方が扱いやすくなります。

実装にはPython標準ライブラリのsqlite3とhttp.serverを使いました。外部パッケージを追加せずに、CSV取込・DB保存・HTTPサーバー・JSONレスポンスまでを一通り作っています。

検証では10行のCSVをSQLiteへ取り込み、/errorsでERRORだけを取得し、さらに?system=api&limit=2のようなクエリで絞り込めることを確認しました。API全体では8件のERRORが返り、apiに絞ると最新2件だけを返せました。

なお、Python公式ドキュメントではhttp.serverは本番利用向けではないと明記されています。今回はローカル検証用の題材として使い、本番環境でそのまま公開しない前提で進めています。

前回のCSV集計から、何をレベルアップしたのか
今回作った構成
検証に使ったCSV
SQLiteのテーブルを作る
CSVをSQLiteへ取り込む
ERRORだけを検索するSQL
HTTP APIを作る
ローカルで起動する
実際にAPIへアクセスして確認した結果
API化すると何が便利になるのか
ただしhttp.serverを本番公開してはいけない
SQLiteにも向いている規模と向いていない規模がある
実務で使うなら追加したい機能
フリーランスSEが試すならダミーデータから始めたい
この「作ってみた」が向いている人
まとめ：CSV集計の次は「保存して検索できる形」にすると学びが広がる
参考情報

前回のCSV集計から、何をレベルアップしたのか

前回の記事では、CSVを読み込んでERROR件数を集計し、システム別・メッセージ別に表示しました。これは定型作業の自動化としては十分実用的ですが、データ自体は処理が終われば残りません。

今回はそこに「永続化」と「検索API」を追加しました。具体的には、CSVをSQLiteのテーブルへ取り込み、あとからHTTPリクエストで検索できるようにしています。

前回の記事はこちらです。

PythonでCSVログのエラー集計ツールを作ってみた｜SEの定型作業を小さく自動化

同じ障害ログを題材にしていますが、今回の中心は「集計」ではなく「保存・検索・API化」です。検索意図も、PythonでCSVを集計する方法ではなく、SQLiteを使った小さなログ検索基盤をどう作るかにあります。

今回作った構成

構成はシンプルです。

CSVからログを読み込む
SQLiteのlogsテーブルへ保存する
systemとlevel、timestampへインデックスを付ける
ローカルHTTPサーバーを起動する
/errorsへアクセスされたらSQLiteからERRORを検索する
検索結果をJSONで返す

外部サービスやクラウドDBは使っていません。1台のPC上で完結するため、仕組みを理解するための検証として扱いやすい構成です。

SQLiteはPython標準ライブラリのsqlite3から利用できます。別途DBサーバーを立てる必要がなく、1つのファイルとして保存できるため、小規模な検証やツール内のデータ保存と相性があります。

HTTP部分はThreadingHTTPServerとBaseHTTPRequestHandlerを利用しました。複数のHTTPリクエストを処理できる形にはしていますが、ここはあくまでローカル検証用です。

検証に使ったCSV

今回は10行のダミーデータを用意しました。INFOやWARNも含めています。

timestamp,system,level,message
2026-09-08 09:01,billing,ERROR,timeout
2026-09-08 09:03,billing,WARN,slow response
2026-09-08 09:05,auth,ERROR,invalid token
2026-09-08 09:10,billing,ERROR,timeout
2026-09-08 09:12,auth,INFO,login success
2026-09-08 09:15,api,ERROR,connection reset
2026-09-08 09:18,api,ERROR,timeout
2026-09-08 09:20,billing,ERROR,db lock
2026-09-08 09:22,api,ERROR,rate limit
2026-09-08 09:30,auth,ERROR,invalid token

ERRORは8件です。billingが3件、apiが3件、authが2件あります。

このデータを直接毎回読み込むのではなく、一度SQLiteへ保存してからAPIで検索します。

SQLiteのテーブルを作る

まず、ログ保存用のテーブルを作ります。

def init_db():
    with sqlite3.connect(DB) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS logs (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp TEXT NOT NULL,
                system TEXT NOT NULL,
                level TEXT NOT NULL,
                message TEXT NOT NULL
            )
        """)
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_logs_system_level "
            "ON logs(system, level)"
        )
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_logs_timestamp "
            "ON logs(timestamp)"
        )

今回は検索条件としてsystemとlevelを使うので、その組み合わせにインデックスを作りました。また、新しいログから並べるためtimestampにもインデックスを付けています。

小規模なデータならインデックスがなくても十分速い場合があります。ただ、API化まで考えるなら、どの列で検索するかを意識しておくことは重要です。データ件数が増えたとき、毎回全件を走査する設計は負荷が上がりやすくなります。

一方で、インデックスは増やせばよいものでもありません。書き込み時にはインデックス更新のコストがかかります。実際の運用では検索条件と書き込み頻度を見ながら決める必要があります。

CSVをSQLiteへ取り込む

CSV取込処理は次のようにしました。

def import_csv(path):
    with sqlite3.connect(DB) as conn, open(
        path, encoding="utf-8", newline=""
    ) as f:
        reader = csv.DictReader(f)
        required = {"timestamp", "system", "level", "message"}

        if not required.issubset(reader.fieldnames or []):
            raise ValueError("CSV columns are incomplete")

        conn.executemany(
            """
            INSERT INTO logs(timestamp, system, level, message)
            VALUES (?, ?, ?, ?)
            """,
            [
                (
                    r["timestamp"],
                    r["system"],
                    r["level"].upper(),
                    r["message"],
                )
                for r in reader
            ],
        )

今回はexecutemany()で複数行をまとめて登録しています。

SQLへ値を渡す部分では、文字列連結ではなく?のプレースホルダを使っています。Python公式のsqlite3ドキュメントでも、Pythonの文字列操作でSQLを組み立てる方法はSQLインジェクションの原因になるため、パラメータ置換を使うよう案内されています。

ローカルツールだから安全対策は不要、とは考えない方がよいです。最初は自分だけが使うツールでも、後から社内共有されたり、別の入力データを受け付けたりすることがあります。最初から安全な書き方を採用しておく方が修正は少なく済みます。

ERRORだけを検索するSQL

APIから呼び出す検索処理は、次のようにしました。

def query_errors(system=None, limit=20):
    sql = """
        SELECT timestamp, system, level, message
        FROM logs
        WHERE level = 'ERROR'
    """
    params = []

    if system:
        sql += " AND system = ?"
        params.append(system)

    sql += " ORDER BY timestamp DESC LIMIT ?"
    params.append(limit)

    with sqlite3.connect(DB) as conn:
        conn.row_factory = sqlite3.Row
        return [dict(row) for row in conn.execute(sql, params)]

systemが指定された場合だけ条件を追加し、limit件まで新しい順に返します。

ポイントは、systemの値をSQL文字列へ直接埋め込んでいないことです。今回の実装ではsystemに' OR 1=1 --のような文字列を渡しても、SQL構文として解釈されず、単なる検索文字列として扱われます。

実際にその値でHTTPリクエストを送りましたが、全件が返ることはなく、該当データ0件という結果になりました。今回のコードだけで完全なセキュリティ対策になるわけではありませんが、少なくともSQL値の渡し方は文字列連結より安全な形にしています。

HTTP APIを作る

HTTP部分は次のように実装しました。

class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        parsed = urlparse(self.path)

        if parsed.path != "/errors":
            self.send_response(404)
            self.end_headers()
            return

        qs = parse_qs(parsed.query)
        system = qs.get("system", [None])[0]

        try:
            limit = min(
                max(int(qs.get("limit", ["20"])[0]), 1),
                100,
            )
        except ValueError:
            limit = 20

        body = json.dumps(
            {"items": query_errors(system, limit)},
            ensure_ascii=False,
        ).encode("utf-8")

        self.send_response(200)
        self.send_header(
            "Content-Type",
            "application/json; charset=utf-8",
        )
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

エンドポイントは/errorsだけです。存在しないパスなら404を返します。

limitは1〜100に制限しました。APIを作るとき、利用者が任意の大きな件数を指定できると、意図せず大量データを返してしまうことがあります。検証用でも上限を設けておくと、API設計の基本を意識しやすくなります。

数値に変換できない値が来た場合は20件へ戻すようにしています。本番APIなら400 Bad Requestを返す方が設計として明確な場合もありますが、今回はコードを複雑にしすぎないためフォールバックにしました。

ローカルで起動する

サーバー起動部分は次の通りです。

host, port = "127.0.0.1", 8765
ThreadingHTTPServer((host, port), Handler).serve_forever()

127.0.0.1へバインドしているため、今回の設定では同じPCからアクセスする前提です。

実行手順は次の2段階です。

python log_api.py import logs.csv
python log_api.py serve

1つ目でCSVをSQLiteへ登録し、2つ目でAPIサーバーを起動します。

実運用で繰り返しCSVを取り込む場合、このままだと同じCSVを再登録すると重複データが入ります。今回はAPIの基本部分を確認するため、重複排除までは実装していません。業務利用へ近づけるなら、一意キーや取込済みファイル管理を追加すべきポイントです。

実際にAPIへアクセスして確認した結果

まず、条件なしで/errorsへアクセスしました。

GET http://127.0.0.1:8765/errors

HTTPステータスは200で、ERRORの8件が新しい順に返りました。先頭は09:30のauth、続いて09:22のapi、09:20のbillingという順番です。

次にapiだけに絞り、2件に制限しました。

GET http://127.0.0.1:8765/errors?system=api&limit=2

結果は次の2件でした。

{
  "items": [
    {
      "timestamp": "2026-09-08 09:22",
      "system": "api",
      "level": "ERROR",
      "message": "rate limit"
    },
    {
      "timestamp": "2026-09-08 09:18",
      "system": "api",
      "level": "ERROR",
      "message": "timeout"
    }
  ]
}

billingを指定した場合は3件が返り、09:20のdb lock、09:10のtimeout、09:01のtimeoutという結果になりました。

存在しない/not-foundへアクセスした場合は404になりました。また、limit=abcを指定した場合は既定値20として処理され、authの2件が返ることを確認しました。

初回の起動確認では接続できないケースがありましたが、その時点では成功扱いにせず、プロセスが起動していることを再確認してから再試行しました。再確認後はHTTP 200でレスポンスを取得できています。

API化すると何が便利になるのか

CLIだけでも集計はできますが、APIにすると他のツールから呼び出しやすくなります。

たとえば、社内ダッシュボードのバックエンドから呼ぶ、別のPythonスクリプトからHTTPで取得する、ローカルの簡易画面から検索するといった拡張ができます。

重要なのは、ログ保存部分と表示部分を分けられることです。SQLiteへデータを入れる処理と、検索結果を使う側がHTTPで分離されるため、後から画面や別処理を追加しやすくなります。

SE実務では、1本の巨大スクリプトへ何でも詰め込むより、役割を分けると変更範囲を小さくできます。CSV形式が変わったなら取込処理、検索条件が変わったならSQL、表示方法を変えたいならAPI利用側、と切り分けやすくなります。

ただしhttp.serverを本番公開してはいけない

ここは今回の検証で最も重要な注意点です。

Python公式ドキュメントではhttp.serverについて、本番利用を推奨しないと明記されています。基本的なセキュリティチェックしか実装していないためです。

今回のコードには認証、認可、TLS、アクセスログ管理、レート制限、監査、CORS制御など、業務向けAPIで検討すべき要素が入っていません。

そのため、インターネットへ公開したり、社内ネットワークだから大丈夫と判断してそのまま常駐させたりするのは避けるべきです。

業務用に発展させるなら、FastAPIやFlaskなどのWebフレームワーク、認証基盤、リバースプロキシ、TLS、監視などを含めて設計する必要があります。今回の目的はHTTPとDBのつながりを手元で理解することです。

SQLiteにも向いている規模と向いていない規模がある

SQLiteは導入しやすい一方、すべてのログ基盤に向くわけではありません。

ローカルで数千〜数十万件程度を検索する小さなツールなら扱いやすい場面がありますが、複数サーバーから大量ログをリアルタイムに集約したり、多数ユーザーが同時に書き込みを行ったりする用途では、専用のログ管理基盤や別のDBを検討する方が自然です。

特に障害ログは増え続けます。保存期間を決めずに取り込み続けると、DBファイルは大きくなります。古いデータをいつ削除するか、バックアップをどうするか、障害時にDB自体をどう復旧するかまで考えると、単なるスクリプトから運用設計の話へ変わってきます。

この「作った後の運用まで考える」部分が、前回のCSV集計より一段レベルが上がるポイントだと感じました。

実務で使うなら追加したい機能

今回のコードを次の段階へ進めるなら、まず重複取込対策が必要です。たとえばtimestamp、system、level、messageの組み合わせへ一意制約を付ける、ファイルハッシュを保存する、といった方法があります。

次に、検索条件を増やしたいところです。開始日時・終了日時、メッセージ部分一致、ERROR以外のレベルなどを指定できると、障害調査で使いやすくなります。

認証も重要です。ログは機密情報を含む場合があるため、APIへアクセスできる利用者を制限する必要があります。APIキーだけで十分なのか、社内SSOと連携するのかは利用範囲によって変わります。

さらに、ログのメッセージをそのまま返す前に、個人情報やトークンをマスクする仕組みも検討が必要です。便利な検索APIを作っても、機密情報を広げる結果になれば本末転倒です。

監視の観点では、API自体のエラー率や応答時間、DBサイズ、取込件数も見たいところです。ツールを作ると、そのツール自身をどう監視するかという新しい運用課題も発生します。

フリーランスSEが試すならダミーデータから始めたい

案件で扱うログをそのまま自分のPCへコピーして試すのは避けた方がよいです。ログには顧客情報、内部IP、URL、アクセストークン、ユーザーIDなどが混在する可能性があります。

まずは今回のようなダミーデータで、CSV取込・DB保存・API検索の流れを理解する。その後、案件環境で利用するなら、データ保存場所、実行端末、アクセス権、ログ保持期間、ソフトウェア利用ルールを確認する方が安全です。

フリーランスSEの場合は、案件をまたいで自作ツールを再利用するときも注意が必要です。コード自体は再利用できても、設定ファイルやログサンプルに前案件の情報が残っていないかを確認する必要があります。

この「作ってみた」が向いている人

今回の題材は、Pythonの基本構文は分かるものの、もう少し実務に近いものを作りたい人に向いています。

CSVを読むだけではなく、DB設計、SQL、インデックス、HTTP、JSON、入力値チェックまで触れるため、小さなバックエンド開発の要素を一通り体験できます。

一方で、初めてPythonに触る人には少し要素が多いかもしれません。その場合は、前回のCSV集計から始めて、SQLite保存、検索関数、API化の順に分けて進めると理解しやすくなります。

まとめ：CSV集計の次は「保存して検索できる形」にすると学びが広がる

今回は、Python標準ライブラリだけで、CSVログをSQLiteへ保存し、HTTP APIからERRORログを検索する仕組みを作りました。

実際の検証では10行を取り込み、ERROR8件を取得できました。system=api&limit=2ではapiの最新2件だけを取得でき、billing指定では3件が返りました。存在しないパスは404、SQLインジェクションを意識した文字列は検索値として扱われ、全件取得にはなりませんでした。

前回の単純集計に比べると、DBへの永続化、SQL設計、インデックス、HTTP API、入力値制御まで考える必要があります。小さな題材ですが、バックエンド開発で必要になる考え方が一気に増えます。

ただし、今回使ったhttp.serverは本番用途向けではありません。業務利用へ進める場合は、認証、TLS、権限制御、ログマスキング、重複取込、バックアップ、監視などを含めて別途設計する必要があります。

「少し難しいものを作ってみたい」という段階なら、完成度の高い大規模システムをいきなり目指すより、CSV→SQLite→APIという流れを1本つなげる方が、各技術の役割を理解しやすい題材になります。

参考情報

標準ライブラリの仕様や推奨事項は更新される可能性があります。実際に利用するPythonバージョンの公式ドキュメントを確認してください。