Python と Cloud Firestore で更新日時を基準に差分を取得しデータを同期する

【概要】

Cloud Firestore のドキュメントに作成日時（created_at）フィールドを持たせ，この値を基準にして，前回の取得時刻より後に作成されたドキュメントだけを取り出す。全件ではなく差分だけを取得することで，取得するデータ量を抑えながら手元のデータを更新する方法を，Python のコード例を交えて説明する。

【目次】

【関連する外部ページ】

【サイト内の関連ページ】

Google Firebase 活用ガイド

前準備

Python 3.12 のインストール

Pythonのインストールを行い、Pythonのプログラムを実行する環境を整える。扱う環境は、Windows搭載パソコンである。金子研究室では、Python 3.12.10を推奨する。

[Windows での Python 3.12 のインストール手順を見るには、ここをクリック]

Windows での Python 3.12 のインストール

以下のいずれかの方法でPython 3.12をインストールする。Pythonがインストール済みの場合、この手順は不要である。

方法 1：winget によるインストール

【インストールコマンドの実行方法】

管理者権限でコマンドプロンプトを起動する（手順：Windowsキーまたはスタートメニュー → cmd と入力 → 右クリック → 「管理者として実行」）。そして、コマンド全体をコマンドプロンプトにコピー＆ペーストする。

--scope machine を指定することで、システム全体（全ユーザー向け）にインストールされる。このオプションの実行には管理者権限が必要である。インストール完了後、コマンドプロンプトを再起動するとPATHが反映される。

REM Python 3.12 をシステム領域にインストール
winget install --id Python.Python.3.12 -e --scope machine --silent --accept-source-agreements --accept-package-agreements --override "/quiet InstallAllUsers=1 PrependPath=1 Include_test=0 Include_pip=1 Include_launcher=1 InstallLauncherAllUsers=1 TargetDir=\"C:\Program Files\Python312\""

REM Python と Scripts を PATH 先頭に追加
powershell -NoProfile -Command "$p='C:\Program Files\Python312'; $s=\"$p\Scripts\"; $c=[Environment]::GetEnvironmentVariable('Path','Machine'); if((Test-Path $p) -and (';'+$c+';' -notlike \"*;$p;*\") -and (';'+$c+';' -notlike \"*;$s;*\")){[Environment]::SetEnvironmentVariable('Path',\"$p;$s;$c\",'Machine')}"

方法 2：インストーラーによるインストール

  1. Python公式サイト（https://www.python.org/downloads/）にアクセスし、「Download Python 3.x.x」ボタンからWindows用インストーラーをダウンロードする。
  2. ダウンロードしたインストーラーを実行する。
  3. 初期画面の下部に表示される「Add python.exe to PATH」にチェックを入れてから「Customize installation」を選択する。このチェックを入れ忘れると、コマンドプロンプトから python コマンドを実行できない。
  4. 「Install Python 3.xx for all users」にチェックを入れ、「Install」をクリックする。

インストールの確認

コマンドプロンプトで以下を実行する。

python --version

バージョン番号（例：Python 3.12.x）が表示されればインストール成功である。「'python' は、内部コマンドまたは外部コマンドとして認識されていません。」と表示される場合は、インストールが正常に完了していない。

Python の開発環境 Visual Studio Code のインストールと Python 用の設定

Python の開発環境Visual Studio Code（プログラムを編集するソフトウェア。以下、VS Code）を整える。

[Windows での Visual Studio Code のインストールと Python 用の設定手順を見るには、ここをクリック]

Windows での Visual Studio Code のインストールと Python 用の設定手順

1. VS Code と拡張機能のインストール

以下のコマンドにより，既存の VS Code を削除し，全ユーザー共有の設定で再インストールしたうえで，拡張機能（VS Code に機能を追加するソフトウェア）をまとめて導入する．

【インストールコマンドの実行方法】

管理者権限でコマンドプロンプトを起動する（手順：Windows キーまたはスタートメニュー → cmd と入力 → 右クリック → 「管理者として実行」）。そして，コマンド全体をコマンドプロンプトにコピー＆ペーストする。

インストールコマンド


REM ============================================================
REM Microsoft Visual Studio Code
REM ============================================================
winget uninstall -e --id Microsoft.VisualStudioCode --silent --disable-interactivity --accept-source-agreements
rmdir /s /q C:\ProgramData\vscode-extensions 2>nul
rmdir /s /q "%APPDATA%\Code" 2>nul
rmdir /s /q "%USERPROFILE%\.vscode" 2>nul
rmdir /s /q "%LOCALAPPDATA%\Microsoft\vscode-update" 2>nul

REM VS Code をシステム領域に新規インストール
winget install --scope machine --id Microsoft.VisualStudioCode -e --silent --accept-source-agreements --accept-package-agreements

REM 全ユーザー共有の拡張機能フォルダ
mkdir C:\ProgramData\vscode-extensions 2>nul
icacls "C:\ProgramData\vscode-extensions" /grant "Everyone:(OI)(CI)M" /T

REM スタートメニューのショートカットを --extensions-dir 付きで再作成
rmdir /s /q "C:\ProgramData\Microsoft\Windows\Start Menu\Programs\Visual Studio Code" 2>nul
del "C:\ProgramData\Microsoft\Windows\Start Menu\Programs\Visual Studio Code.lnk" 2>nul
powershell -NoProfile -Command "$s=New-Object -ComObject WScript.Shell; $lnk=$s.CreateShortcut('C:\ProgramData\Microsoft\Windows\Start Menu\Programs\Visual Studio Code.lnk'); $lnk.TargetPath='C:\Program Files\Microsoft VS Code\Code.exe'; $lnk.Arguments='--extensions-dir \"C:\ProgramData\vscode-extensions\"'; $lnk.Save()"
REM ショートカットの検証
powershell -NoProfile -Command "$s=New-Object -ComObject WScript.Shell; $lnk=$s.CreateShortcut('C:\ProgramData\Microsoft\Windows\Start Menu\Programs\Visual Studio Code.lnk'); Write-Host 'TargetPath:' $lnk.TargetPath; Write-Host 'Arguments:' $lnk.Arguments"

REM ファイル / フォルダ右クリックの「Code で開く」を登録
reg add "HKLM\SOFTWARE\Classes\*\shell\VSCode\command" /ve /d "\"C:\Program Files\Microsoft VS Code\Code.exe\" --extensions-dir \"C:\ProgramData\vscode-extensions\" \"%1\"" /f
reg add "HKLM\SOFTWARE\Classes\Directory\shell\VSCode\command" /ve /d "\"C:\Program Files\Microsoft VS Code\Code.exe\" --extensions-dir \"C:\ProgramData\vscode-extensions\" \"%1\"" /f
reg add "HKLM\SOFTWARE\Classes\Directory\Background\shell\VSCode\command" /ve /d "\"C:\Program Files\Microsoft VS Code\Code.exe\" --extensions-dir \"C:\ProgramData\vscode-extensions\" \"%V\"" /f

REM --extensions-dir 付きで起動する code.cmd ラッパを作成
REM （%* を echo で書くと対話的 cmd で失われるため、PowerShell で [char]37+'*' を書き出す）
powershell -NoProfile -Command "$pct=[char]37; $q=[char]34; $c='@echo off'+[char]13+[char]10+$q+'C:\Program Files\Microsoft VS Code\bin\code.cmd'+$q+' --extensions-dir '+$q+'C:\ProgramData\vscode-extensions'+$q+' '+$pct+'*'+[char]13+[char]10; [IO.File]::WriteAllText('C:\ProgramData\vscode-extensions\vscode.cmd',$c,[Text.Encoding]::ASCII)"

REM 拡張機能のインストール
set "CODE=C:\Program Files\Microsoft VS Code\bin\code.cmd"
"%CODE%" --extensions-dir "C:\ProgramData\vscode-extensions" --uninstall-extension GitHub.copilot
"%CODE%" --extensions-dir "C:\ProgramData\vscode-extensions" --uninstall-extension GitHub.copilot-chat
"%CODE%" --extensions-dir "C:\ProgramData\vscode-extensions" --install-extension ms-python.python
"%CODE%" --extensions-dir "C:\ProgramData\vscode-extensions" --install-extension ms-python.vscode-pylance
"%CODE%" --extensions-dir "C:\ProgramData\vscode-extensions" --install-extension ms-python.debugpy
"%CODE%" --extensions-dir "C:\ProgramData\vscode-extensions" --install-extension MS-CEINTL.vscode-language-pack-ja
"%CODE%" --extensions-dir "C:\ProgramData\vscode-extensions" --install-extension saoudrizwan.claude-dev
"%CODE%" --extensions-dir "C:\ProgramData\vscode-extensions" --install-extension rust-lang.rust-analyzer
"%CODE%" --extensions-dir "C:\ProgramData\vscode-extensions" --install-extension tamasfe.even-better-toml
"%CODE%" --extensions-dir "C:\ProgramData\vscode-extensions" --install-extension anthropic.claude-code
"%CODE%" --extensions-dir "C:\ProgramData\vscode-extensions" --install-extension almenon.arepl
"%CODE%" --extensions-dir "C:\ProgramData\vscode-extensions" --list-extensions --show-versions
echo === セットアップ完了 ===

2. Python インタプリタの選択

同一マシンに複数の Python がインストールされている場合，VS Code で使用する Python 本体（インタプリタ：Python プログラムを解釈・実行するソフトウェア）を選択する必要がある．

  1. コマンドパレット（コマンド名で機能を呼び出す VS Code の入力欄）を開く（Ctrl+Shift+P）
  2. Python: Select Interpreter と入力する
  3. 表示される一覧から，使用する Python（例：C:\Program Files\Python312\python.exe）を選択する．

Python プログラム実行手順

[Windows での Python プログラム実行手順を見るには、ここをクリック]

Windows での Python 実行手順（Visual Studio Codeを使用）

プログラムファイルの作成と保存

  1. 左サイドバーの「エクスプローラー」アイコン（Ctrl+Shift+E）をクリックする
  2. 「NO FOLDER OPENED」（作業対象フォルダが未選択の状態）と表示される場合は，「Open Folder」をクリックし，プログラムを保存するフォルダを選択する

    続いて「フォルダを信用するか」を確認する画面（フォルダ内のコードを実行してよいか確認する VS Code の仕組み）が表示されるので，チェックして Yes を選択する

  3. フォルダ名の右側に表示される「新しいファイル」アイコンをクリックする
  4. ファイル名（例：aitask.py．ファイル名は何でも良い）を入力し Enter を押す．拡張子は .py（Python ファイルを示す拡張子）とする
  5. 実行したいコードを選択し，Ctrl+C でコピーする．VS Code のエディタ領域に Ctrl+V で貼り付ける
  6. Ctrl+S で保存する

プログラムの実行

  1. エディタ右上の三角形「▷」アイコン（Run Python File：現在開いている Python ファイルを実行するボタン）をクリックする．または，エディタ上で右クリックし「ターミナルで Python ファイルを実行」を選択する
  2. VS Code 下部のターミナル（コマンドの入出力を表示する画面）に，実行結果（print 関数の出力等）が表示される
  3. tkinter（Python 標準の GUI ライブラリ）のファイル選択ダイアログを使うプログラムを実行した場合は，ダイアログが開くので対象画像を選択する
  4. VS Code 下部のターミナルで実行結果を確認する．OpenCV ウィンドウ（OpenCV が画像を表示するために開く専用ウィンドウ）が開いた場合はそちらも確認する．OpenCV ウィンドウは，マウスクリックでウィンドウをアクティブ（操作対象の状態）にしてからキーを押すと終了する

Cloud Firestore への接続と差分取得

  1. 初期設定
    • Firebase公式サイト： https://firebase.google.com/?hl=ja にアクセスし， プロジェクトを作成して認証用の秘密鍵をダウンロードする。
    • Firebase Admin SDK（firebase-admin パッケージ）をインストールする。この SDK により，サーバ側から Cloud Firestore へ直接アクセスできる。 
      pip install -U --no-user firebase-admin

    * 秘密鍵の取得やプロジェクト作成の詳細な手順は「Google Firebase の Cloud Firestore 基本操作ガイド」を参照する。

  2. データモデルの設計

    各ドキュメントに created_at フィールド（datetime 型）を持たせる。このフィールドはドキュメントの作成時刻を表し，差分取得の基準として使う。新しく作成したドキュメントには，それまでのドキュメントより後の時刻が入るため，この時刻を基準にすると，前回より後に作成されたドキュメントだけを取り出せる。

  3. 差分の取得

    秘密鍵の JSON ファイル（serviceAccountKey.json）のパスは実際のパスに置き換える。

    基準となる時刻（datetime.datetime(2019, 4, 8, 9, 30, 0)）より後に作成されたドキュメントだけを取得する。where には FieldFilter（フィールドの条件を表すオブジェクト。created_at が基準時刻より後、という条件を指定している）を渡す。 

    import datetime
import firebase_admin
from firebase_admin import credentials
from firebase_admin import firestore
from google.cloud.firestore_v1.base_query import FieldFilter

cred = credentials.Certificate('serviceAccountKey.json')
firebase_admin.initialize_app(cred)

db = firestore.client()
ref = db.collection('docs')

a = ref.where(filter=FieldFilter('created_at', '>', datetime.datetime(2019, 4, 8, 9, 30, 0)))
docs = a.stream()
for doc in docs:
    print('{} => {}'.format(doc.id, doc.to_dict()))
  4. 取得したデータの中の最新時刻を求める

    取得したドキュメントの中から，最も新しい created_at を求め，次回の取得の基準として使う。基準時刻には，協定世界時（UTC）を明示した datetime を使う。 

    import datetime
import firebase_admin
from firebase_admin import credentials
from firebase_admin import firestore
from google.cloud.firestore_v1.base_query import FieldFilter

cred = credentials.Certificate('serviceAccountKey.json')
firebase_admin.initialize_app(cred)

db = firestore.client()
ref = db.collection('docs')

# 基準時刻の設定
last = datetime.datetime(2019, 4, 8, 0, 30, 0, 0, datetime.timezone.utc)
a = ref.where(filter=FieldFilter('created_at', '>', last))

docs = a.stream()
new_last = last
for doc in docs:
    print('{} => {}'.format(doc.id, doc.to_dict()))
    if new_last < doc.to_dict()['created_at']:
        new_last = doc.to_dict()['created_at']
  5. 定期的に差分を取得する

    5秒間隔で差分の取得を繰り返す。この間隔は用途に応じて変更してよい。全件を取得せず，前回取得した最新時刻を基準に差分だけを取得する。基準時刻を進めるときは10秒のマージンを引くことで，時刻のずれによる取りこぼしを防ぐ。

    【関連情報】 定期実行処理の実装ガイド 

    import time
import threading

import datetime
import firebase_admin
from firebase_admin import credentials
from firebase_admin import firestore
from google.cloud.firestore_v1.base_query import FieldFilter

cred = credentials.Certificate('serviceAccountKey.json')
firebase_admin.initialize_app(cred)

db = firestore.client()
ref = db.collection('docs')

base = datetime.datetime(2019, 4, 8, 0, 30, 0, 0, datetime.timezone.utc)
last = base
cache = {}

def worker():
    global last
    global cache
    a = ref.where(filter=FieldFilter('created_at', '>', last))

    docs = a.stream()
    new_last = last
    for doc in docs:
        cache[doc.id] = doc.to_dict()
        if new_last < doc.to_dict()['created_at']:
            new_last = doc.to_dict()['created_at']
    print("cache data")
    print(cache)
    if last < (new_last - datetime.timedelta(0, 10)):
        last = new_last - datetime.timedelta(0, 10)

def mainloop(time_interval, f):
    now = time.time()
    while True:
        t = threading.Thread(target=f)
        t.start()
        t.join()
        wait_time = time_interval - ((time.time() - now) % time_interval)
        time.sleep(wait_time)

mainloop(5, worker)

差分更新の仕組み

データベース構造

各ドキュメントに，次のフィールドを持たせる。created_at を基準にすることで，差分の取得ができる。 

{
    "id": "document_id",
    "created_at": timestamp,  // ドキュメント作成日時
    "updated_at": timestamp,  // 最終更新日時
    "content": {
        // ドキュメントの実データ
    }
}

差分更新処理の流れ

  1. タイムスタンプの管理
    created_at フィールドはドキュメントの作成時に設定する。このフィールドを基準にして差分を取得する。
  2. データの取得

    前回の取得時刻より後に作成されたドキュメントだけを取得する。これにより，取得するデータ量を抑えられる。 

    def fetch_updates(last_sync_time):
    updates = ref.where(filter=FieldFilter('created_at', '>', last_sync_time))
    return updates.stream()
  3. ローカルキャッシュの管理

    取得したデータをローカルのキャッシュ（手元のメモリ上に保存したデータ）に入れ，次回の取得時に参照する。 

    def update_cache(cache, new_docs):
    for doc in new_docs:
        cache[doc.id] = doc.to_dict()
    return cache

運用上の注意点