diff --git a/.devcontainer/dev.compose.yaml b/.devcontainer/dev.compose.yaml deleted file mode 100644 index 7478122..0000000 --- a/.devcontainer/dev.compose.yaml +++ /dev/null @@ -1,6 +0,0 @@ -services: - dev: - image: mcr.microsoft.com/devcontainers/base:debian - volumes: - - .:/workspace - command: sleep infinity diff --git a/.devcontainer/devcontainer.json b/.devcontainer/devcontainer.json index e03ff8f..bfdfa1e 100644 --- a/.devcontainer/devcontainer.json +++ b/.devcontainer/devcontainer.json @@ -1,30 +1,41 @@ { "name": "weko-group-cache-db Dev Container", - "dockerComposeFile": [ - "../compose.yaml", - "dev.compose.yaml" - ], - "runServices": ["dev", "redis"], - "service": "dev", + "image": "mcr.microsoft.com/devcontainers/base:bookworm", "remoteUser": "vscode", - "workspaceFolder": "/workspace", + + "initializeCommand": "docker network create weko-group-cache-db_default || true", + "runArgs": [ + "--network", + "weko-group-cache-db_default" + ], "features": { "ghcr.io/devcontainers/features/python:1": { "version": "3.14" + }, + "ghcr.io/devcontainers/features/docker-outside-of-docker:1": { + "version": "latest", + "dockerDashComposeVersion": "latest", + "enableNonRootDocker": true, + "moby": false } }, "remoteEnv": { "VIRTUAL_ENV": "/home/vscode/.venv", "UV_PROJECT_ENVIRONMENT": "/home/vscode/.venv" }, - "postCreateCommand": "chmod +x .devcontainer/setup.sh && .devcontainer/setup.sh vscode", + "mounts": [ + "source=/var/run/docker.sock,target=/var/run/docker.sock,type=bind" + ], + + "onCreateCommand": "/bin/sh .devcontainer/setup.sh install_deps", + "postAttachCommand": "/bin/sh .devcontainer/setup.sh compose_up", "customizations": { "vscode": { "settings": { - "python.pythonPath": "/code/.venv/bin/python", "python.languageServer": "Pylance", + "python.defaultInterpreterPath": "~/.venv/bin/python", "[python]": { "editor.rulers": [88], "editor.formatOnSave": true, @@ -33,7 +44,9 @@ "source.organizeImports": "explicit" }, "editor.defaultFormatter": "charliermarsh.ruff" - } + }, + "python.useEnvironmentsExtension": false, + "python.terminal.activateEnvironment": false }, "extensions": [ "visualstudioexptteam.vscodeintellicode", diff --git a/.devcontainer/setup.sh b/.devcontainer/setup.sh index 048ca33..0e369ac 100755 --- a/.devcontainer/setup.sh +++ b/.devcontainer/setup.sh @@ -1,26 +1,38 @@ #!/bin/bash -set -e -REMOTE_USER=$1 -BASHRC_FILE="/home/${REMOTE_USER}/.bashrc" +install_deps() { + # Install Python dependencies + pip install --upgrade pip && pip install uv + uv sync && uv pip install -e . -if [ -z "$REMOTE_USER" ]; then - echo "Error: User name argument is missing." - exit 1 -fi + if ! grep -q "venv_activate_reload" ~/.bashrc 2>/dev/null; then + { + echo "" + echo "# venv_activate_reload" + echo "if [ -f ~/.venv/bin/activate ]; then" + echo " source ~/.venv/bin/activate" + echo "fi" + } >> ~/.bashrc + fi +} + + +compose_up() { + docker compose up -d --force-recreate redis +} -if ! grep -q "venv_activate_reload" "$BASHRC_FILE" 2>/dev/null; then - { - echo "" - echo "# venv_activate_reload" - echo "if [ -f /home/${REMOTE_USER}/.venv/bin/activate ]; then" - echo " source /home/${REMOTE_USER}/.venv/bin/activate" - echo "fi" - } >> "$BASHRC_FILE" +FUNCTION_NAME=$1 - sudo chown ${REMOTE_USER}:${REMOTE_USER} "$BASHRC_FILE" +if [ -z "$FUNCTION_NAME" ]; then + echo "Error: Function name argument is missing." + exit 1 fi +shift -pip install --upgrade pip -pip install uv && uv sync && uv pip install -e . +if type -t "$FUNCTION_NAME" | grep -q 'function'; then + "$FUNCTION_NAME" "$@" +else + echo "Error: Function '$FUNCTION_NAME' not found." + exit 1 +fi diff --git a/Dockerfile b/Dockerfile index ce47ac2..db62bb5 100644 --- a/Dockerfile +++ b/Dockerfile @@ -27,7 +27,6 @@ FROM base AS production ENV VIRTUAL_ENV="/code/.venv" ENV PATH="$VIRTUAL_ENV/bin:$PATH" -RUN uv sync --no-dev --frozen && \ - uv pip install . +RUN uv sync --no-dev --frozen CMD ["sh", "-c", "COLUMNS=120 exec wgcd run -d /var/mnt -l /var/repositories.txt"] diff --git a/README.md b/README.md index a70014d..c144930 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # weko-group-cache-db / グループ情報キャッシュ機能 -Coverage
Coverage Report
FileStmtsMissCover
weko_group_cache_db
   cli.py520100%
   config.py840100%
   exc.py90100%
   groups.py860100%
   loader.py1110100%
   logger.py250100%
   redis.py260100%
TOTAL3930100%
+

Coverage

Coverage Report
FileStmtsMissBranchBrPartCoverMissing
weko_group_cache_db
   cli.py520120100% 
   config.py74080100% 
   exc.py9000100% 
   groups.py1170180100% 
   loader.py11712411 99%
   logger.py24020100% 
   redis.py26020100% 
   signals.py22000100% 
TOTAL441166199% 
## 概要 @@ -100,7 +100,7 @@ $ wgcd run --help 1つの機関に接続しているグループ情報を取得し、キャッシュDBへ登録する。 マウントされたディレクトリからFQDNリストファイルとSSL証明書・SSL鍵ファイルを使用して実行するディレクトリベースの実行方法と、機関情報を記述したTOMLファイルを使用して実行するファイルベースの実行方法がある。 -### ヘルプ表示 +#### ヘルプ表示 ``` $ wgcd one --help @@ -208,7 +208,6 @@ $ wgcd one --help ### カスタム設定値 [config.toml](./config.toml)に設定値を記載することで、実行時に設定値を反映して実行することが可能。 -また、環境変数、`.env`ファイル、Kubernetesのシークレットマウントによっても設定値を反映することが可能。 `MAP_GROUPS_API_ENDPOINT`以外はデフォルトの設定値があるため、それ以外は必要に応じて上書きしたい設定値のみを記載すればよい。 `config.toml`に記述する場合、大文字小文字は区別されない. @@ -227,7 +226,7 @@ $ wgcd one --help | REQUEST_RETRY_BASE | 数値 | 4 | グループ情報取得・キャッシュDBへの登録処理のリトライ時の指数バックオフの基準時間（秒） | | REQUEST_RETRY_FACTOR | 数値 | 5 | グループ情報取得・キャッシュDBへの登録処理のリトライ時の指数バックオフの係数（秒） | | REQUEST_RETRY_MAX | 数値 | 90 | グループ情報取得・キャッシュDBへの登録処理のリトライ時の指数バックオフの最大時間（秒） | -| REDIS_TYPE | 文字列 | redis | 使用するRedisの種類
"redis", "sentinel"のうちから指定 | +| REDIS_TYPE | 文字列 | RedisCache | 使用するRedisの種類
"RedisCache", "RedisSentinelCache"のうちから指定 | | REDIS_HOST | 文字列 | localhost | Redisのホスト名 | | REDIS_PORT | 数値 | 6379 | Redisのポート番号 | | REDIS_DB_INDEX | 数値 | 4 | グループ情報を格納するRedisのDBインデックス | @@ -239,15 +238,15 @@ $ wgcd one --help ※ Redis Sentinel の設定値をTOMLファイルに記載する場合の例 ``` [[sentinels]] -host = "sendinel-1" +host = "sentinel-1" port = 26379 [[sentinels]] -host = "sendinel-2" +host = "sentinel-2" port = 26378 [[sentinels]] -host = "sendinel-3" +host = "sentinel-3" port = 26377 ``` @@ -255,9 +254,49 @@ port = 26377 1. TOMLファイル 2. 環境変数 -3. `.env`ファイル -4. Kubernetesのシークレットマウント -5. デフォルト値（`MAP_GROUPS_API_ENDPOINT`以外） +3. デフォルト値（`MAP_GROUPS_API_ENDPOINT`以外） + + +## シグナル + +[blinker](https://blinker.readthedocs.io/) のシグナルにより、処理の進捗と各機関の処理結果を外部から受け取ることができる。 + +```python +from weko_group_cache_db.signals import executed_signal, progress_signal + + +@progress_signal.connect +def on_progress(sender, data): + print(f"{data.status}: {data.done}/{data.total}") +``` + +| シグナル | 送信タイミング | 対象コマンド | データ型 | +| ----------------- | --------------------------------------------------------- | -------------- | -------------- | +| `progress_signal` | 各機関の処理開始前、および全機関の処理完了後 | `run` | `ProgressData` | +| `executed_signal` | 各機関の処理の試行ごと（リトライを含む）の成功時・失敗時 | `run`, `one` | `ExecutedData` | + +### progress_signal + +| 項目 | 型 | 説明 | +| --------- | ------ | -------------------------------------------------------------------- | +| `status` | 文字列 | `started`：最初の機関の処理開始前
`in_progress`：2件目以降の機関の処理開始前
`completed`：全機関の処理完了後（失敗した機関があっても送信される） | +| `total` | 数値 | 処理対象の機関数 | +| `done` | 数値 | 処理済みの機関数 | +| `current` | 文字列 | これから処理する機関のFQDN（`completed` の場合は `N/A`） | + +### executed_signal + +| 項目 | 型 | 説明 | +| --------------- | -------- | -------------------------------------------------------------------- | +| `fqdn` | 文字列 | 機関のFQDN | +| `status` | 文字列 | `success`：グループ情報のRedisへの登録に成功
`failed`：グループ情報の取得またはRedisへの登録に失敗 | +| `retries` | 数値 | その試行までのリトライ回数（初回の試行は 0） | +| `error_type` | 文字列 | 失敗時の例外クラス名（成功時は `None`） | +| `error_message` | 文字列 | 失敗時のエラーメッセージ（成功時は `None`） | +| `updated_at` | 日時 | 成功時はRedisへの登録日時、失敗時はシグナルの送信日時（UTC） | + +※ `failed` はリトライのたびに送信される。 + ## 開発環境構築 diff --git a/compose.yaml b/compose.yaml index 7b7faa0..60d86bf 100644 --- a/compose.yaml +++ b/compose.yaml @@ -9,6 +9,8 @@ services: volumes: - .:/code - /code/.venv + networks: + - default redis: restart: "always" @@ -19,3 +21,10 @@ services: resources: limits: memory: 100m + networks: + - default + +networks: + default: + external: true + name: weko-group-cache-db_default diff --git a/config.toml b/config.toml index 089793b..dc83e77 100644 --- a/config.toml +++ b/config.toml @@ -18,7 +18,7 @@ # cache_ttl = 86400 # === Map groups API endpoint. === -map_groups_api_endpoint = "https://sample.gakunin.jp/api/groups/" +map_groups_api_endpoint = "https://sptest.cg.gakunin.jp/api/groups/" # === Request timeout (in seconds) when connecting to mAP API. === # request_timeout = 20 @@ -42,8 +42,8 @@ map_groups_api_endpoint = "https://sample.gakunin.jp/api/groups/" # === Maximum time (in seconds) for exponential backoff during request retries. === # request_retry_max = 90 -# === Redis type to use. `redis` or `sentinel` is allowed. === -# redis_type = "redis" +# === Redis type to use. `RedisCache` or `RedisSentinelCache` is allowed. === +# redis_type = "RedisCache" # === Redis service host name. === # redis_host = "localhost" diff --git a/pyproject.toml b/pyproject.toml index b1581dc..821b9d6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,14 +1,21 @@ +#:schema https://json.schemastore.org/pyproject.json + +[build-system] +requires = ["uv_build>=0.13.5,<0.14"] +build-backend = "uv_build" + [project] name = "weko-group-cache-db" -version = "1.0.0rc4" +version = "1.0.0rc6" description = "CLI tool for caching mAP group information on JAIRO Cloud." readme = "README.md" requires-python = ">=3.14" dependencies = [ "backoff>=2.2.1", + "blinker>=1.9.0", "inflect>=7.5.0", "pydantic-settings>=2.12.0", - "redis>=7.0.1", + "redis>=6.4.0", "requests>=2.32.5", "rich-click>=1.9.4", "werkzeug>=3.1.3", @@ -29,6 +36,10 @@ dev = [ "toml>=0.10.2", ] +[tool.uv.build-backend] +module-name = "weko_group_cache_db" +module-root = "" + [tool.taskipy.tasks] # refer to https://github.com/taskipy/taskipy?tab=readme-ov-file#usage test = "pytest" @@ -46,17 +57,22 @@ addopts = [ [tool.coverage.run] source = ["weko_group_cache_db"] -omit = ["*/__init__.py", "*/tests/*"] + +[tool.coverage.report] +omit = ["__init__.py", "tests/*"] +exclude_also = [ + "if t.TYPE_CHECKING:", +] [tool.ruff.lint] # refer to https://docs.astral.sh/ruff/settings/#lint ignore = [ - "D203", - "D213", - "TC006", - "COM812", - "ANN201", - "TRY400", + "incorrect-blank-line-before-class", + "multi-line-summary-second-line", + "runtime-cast-value", + "missing-trailing-comma", + "missing-return-type-undocumented-public-function", + "error-instead-of-exception", ] preview = true select = ["ALL"] diff --git a/tests/conftest.py b/tests/conftest.py index 5c65837..84fcd06 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -51,7 +51,7 @@ def row_config(): "REQUEST_RETRY_BASE": 2, "REQUEST_RETRY_FACTOR": 15, "REQUEST_RETRY_MAX": 60, - "REDIS_TYPE": "sentinel", + "REDIS_TYPE": "RedisSentinelCache", "REDIS_HOST": "redis", "REDIS_PORT": 26379, "REDIS_DB_INDEX": 2, diff --git a/tests/ruff.toml b/tests/ruff.toml index bdb096a..81d25b4 100644 --- a/tests/ruff.toml +++ b/tests/ruff.toml @@ -4,14 +4,14 @@ target-version = "py314" [lint] ignore = [ "D", - "S101", + "assert", "ANN", - "COM812", - "TC006", - "DTZ005", - "PLC2701", - "SLF001", - "SIM300", + "missing-trailing-comma", + "runtime-cast-value", + "call-datetime-now-without-tzinfo", + "import-private-name", + "private-member-access", + "yoda-conditions", ] preview = true select = ["ALL"] diff --git a/tests/test_config.py b/tests/test_config.py index b347f1a..7504e46 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -3,10 +3,8 @@ # import typing as t -import pytest import toml -from pydantic_core import ValidationError from werkzeug.local import LocalProxy from weko_group_cache_db.config import Settings, _current_config, config, setup_config @@ -37,16 +35,6 @@ def test_settings_default_values(): assert settings.REDIS_URL == "redis://localhost:6379/4" -def test_settings_missing_required_field(): - with pytest.raises(ValidationError) as excinfo: - Settings() # pyright: ignore[reportCallIssue] - - errors = excinfo.value.errors() - assert len(errors) == 1 - assert errors[0]["loc"] == ("MAP_GROUPS_API_ENDPOINT",) - assert errors[0]["msg"] == "Field required" - - def test_setup_config(tmp_path, row_config): config_path = tmp_path / "config.toml" row_config = {key: value for key, value in row_config.items() if key == "MAP_GROUPS_API_ENDPOINT"} @@ -90,11 +78,11 @@ def test_setup_config_overrides_lower_case(tmp_path, row_config): assert config_dict == row_config -def test_setup_config_invalid_path(): - with pytest.raises(ValidationError) as excinfo: - setup_config("non_existent_config.toml") +def test_setup_config_dict(row_config): + setup_config(row_config) + assert t.cast(LocalProxy, config)._get_current_object() is _current_config.get() + - errors = excinfo.value.errors() - assert len(errors) == 1 - assert errors[0]["loc"] == ("MAP_GROUPS_API_ENDPOINT",) - assert errors[0]["msg"] == "Field required" +def test_setup_config_instance(row_config): + setup_config(Settings(**row_config)) + assert t.cast(LocalProxy, config)._get_current_object() is _current_config.get() diff --git a/tests/test_groups.py b/tests/test_groups.py index 859028f..9d4df82 100644 --- a/tests/test_groups.py +++ b/tests/test_groups.py @@ -12,6 +12,7 @@ from weko_group_cache_db.exc import UpdateError from weko_group_cache_db.groups import fetch_all, fetch_and_cache, fetch_map_groups, fetch_one, set_groups_to_redis from weko_group_cache_db.loader import Institution +from weko_group_cache_db.signals import executed_signal, progress_signal def test_fetch_all_success_toml(institutions_data, set_test_config, log_capture): @@ -21,11 +22,13 @@ def test_fetch_all_success_toml(institutions_data, set_test_config, log_capture) set_test_config() mock_store = MagicMock() + mock_receiver = MagicMock() with ( patch("weko_group_cache_db.groups.connection", return_value=mock_store), patch("weko_group_cache_db.groups.load_institutions") as mock_load, patch("weko_group_cache_db.groups.fetch_and_cache") as mock_fetch_and_cache, patch("weko_group_cache_db.groups.time") as mock_time, + progress_signal.connected_to(mock_receiver), ): mock_load.return_value = institutions mock_func = MagicMock(side_effect=[2, 3]) @@ -40,6 +43,13 @@ def test_fetch_all_success_toml(institutions_data, set_test_config, log_capture) assert log_capture.records[0].getMessage() == f"Successfully cached 2 groups for {data[0]['fqdn']}." assert log_capture.records[1].getMessage() == f"Successfully cached 3 groups for {data[1]['fqdn']}." mock_time.sleep.assert_called_once() + assert mock_receiver.call_count == num_institutions + 1 + assert all(call.args == (fetch_all,) for call in mock_receiver.call_args_list) + sent = [call.kwargs["data"] for call in mock_receiver.call_args_list] + assert [d.status for d in sent] == ["started", "in_progress", "completed"] + assert [d.done for d in sent] == [0, 1, 2] + assert [d.current for d in sent] == [data[0]["fqdn"], data[1]["fqdn"], "N/A"] + assert all(d.total == num_institutions for d in sent) def test_fetch_all_success_directory(institutions_data, set_test_config, log_capture): @@ -87,11 +97,13 @@ def test_fetch_all_request_error(institutions_data, set_test_config, log_capture test_config = set_test_config() mock_store = MagicMock() + mock_receiver = MagicMock() with ( patch("weko_group_cache_db.groups.connection", return_value=mock_store), patch("weko_group_cache_db.groups.load_institutions", return_value=institutions), patch("weko_group_cache_db.groups.fetch_and_cache") as mock_fetch_and_cache, patch("weko_group_cache_db.groups.time"), + progress_signal.connected_to(mock_receiver), ): num_groups = 2 mock_func = MagicMock(side_effect=[num_groups, requests.RequestException("Request failed")]) @@ -111,6 +123,10 @@ def test_fetch_all_request_error(institutions_data, set_test_config, log_capture assert isinstance(error_instance, UpdateError) assert isinstance(error_instance.origin, requests.RequestException) assert str(error_instance) == f"FQDN: {data[1]['fqdn']}, Request failed" + sent = [call.kwargs["data"] for call in mock_receiver.call_args_list] + assert [d.status for d in sent] == ["started", "in_progress", "completed"] + assert [d.done for d in sent] == [0, 1, 2] + assert [d.current for d in sent] == [data[0]["fqdn"], data[1]["fqdn"], "N/A"] def test_fetch_all_redis_error(institutions_data, set_test_config, log_capture): @@ -254,13 +270,17 @@ def test_fetch_and_cache_success(institutions_data, set_test_config): institution = Institution(**data[0]) set_test_config() groups = ["jc_group1", "jc_group2"] + updated_at = datetime(2025, 1, 1, tzinfo=UTC) mock_store = MagicMock() + mock_receiver = MagicMock() with ( patch("weko_group_cache_db.groups.fetch_map_groups") as mock_fetch, patch("weko_group_cache_db.groups.set_groups_to_redis") as mock_set, + executed_signal.connected_to(mock_receiver), ): mock_fetch.return_value = groups + mock_set.return_value = updated_at func = fetch_and_cache() count = func(institution, mock_store) @@ -268,6 +288,15 @@ def test_fetch_and_cache_success(institutions_data, set_test_config): assert count == len(groups) mock_fetch.assert_called_once_with(institution) mock_set.assert_called_once_with(institution.fqdn, groups, store=mock_store) + mock_receiver.assert_called_once() + assert mock_receiver.call_args.args == (fetch_all,) + sent = mock_receiver.call_args.kwargs["data"] + assert sent.fqdn == institution.fqdn + assert sent.status == "success" + assert sent.retries == 0 + assert sent.error_type is None + assert sent.error_message is None + assert sent.updated_at == updated_at def test_fetch_and_cache_all_request_exception(institutions_data, set_test_config, log_capture): @@ -276,9 +305,11 @@ def test_fetch_and_cache_all_request_exception(institutions_data, set_test_confi test_config = set_test_config(REQUEST_RETRY_FACTOR=0, REQUEST_RETRIES=2) mock_store = MagicMock() + mock_receiver = MagicMock() with ( patch("weko_group_cache_db.groups.fetch_map_groups") as mock_fetch, patch("weko_group_cache_db.groups.set_groups_to_redis") as mock_set, + executed_signal.connected_to(mock_receiver), ): mock_fetch.side_effect = requests.RequestException("Request failed") @@ -292,6 +323,13 @@ def test_fetch_and_cache_all_request_exception(institutions_data, set_test_confi record.getMessage() == f"Failed to fetch groups from mAP API for institution: {institution.fqdn}" for record in log_capture.records ) + assert mock_receiver.call_count == test_config.REQUEST_RETRIES + 1 + sent = [call.kwargs["data"] for call in mock_receiver.call_args_list] + assert all(d.fqdn == institution.fqdn for d in sent) + assert all(d.status == "failed" for d in sent) + assert [d.retries for d in sent] == [0, 1, 2] + assert all(d.error_type == "RequestException" for d in sent) + assert all(d.error_message == "Request failed" for d in sent) def test_fetch_and_cache_all_redis_error(institutions_data, set_test_config, log_capture): @@ -327,9 +365,11 @@ def test_fetch_and_cache_success_after_retries(institutions_data, set_test_confi groups = ["jc_group1", "jc_group2"] mock_store = MagicMock() + mock_receiver = MagicMock() with ( patch("weko_group_cache_db.groups.fetch_map_groups") as mock_fetch, patch("weko_group_cache_db.groups.set_groups_to_redis") as mock_set, + executed_signal.connected_to(mock_receiver), ): retries_fetch = 2 mock_fetch.side_effect = [requests.RequestException("Request failed"), groups, groups] @@ -346,6 +386,11 @@ def test_fetch_and_cache_success_after_retries(institutions_data, set_test_confi record.getMessage() == f"Failed to cache groups to Redis for institution: {institution.fqdn}" for record in log_capture.records ) + sent = [call.kwargs["data"] for call in mock_receiver.call_args_list] + assert [d.status for d in sent] == ["failed", "failed", "success"] + assert [d.retries for d in sent] == [0, 1, 2] + assert [d.error_type for d in sent] == ["RequestException", "RedisError", None] + assert [d.error_message for d in sent] == ["Request failed", "Cache error", None] def test_fetch_map_groups(institutions_data, set_test_config): @@ -438,3 +483,57 @@ def test_set_groups_to_redis_no_expire(institutions_data, set_test_config): mock_store.hset.assert_called_once_with(redis_key, mapping={"updated_at": timestamp, "groups": ",".join(groups)}) mock_store.persist.assert_called_once_with(redis_key) mock_store.expire.assert_not_called() + + +def test_fetch_all_progress_signal_receiver_error(institutions_data, set_test_config, log_capture): + num_institutions = 2 + data = institutions_data(num_institutions) + institutions = [Institution(**data[0]), Institution(**data[1])] + set_test_config() + + mock_receiver = MagicMock(side_effect=Exception("Receiver error")) + with ( + patch("weko_group_cache_db.groups.connection"), + patch("weko_group_cache_db.groups.load_institutions", return_value=institutions), + patch("weko_group_cache_db.groups.fetch_and_cache") as mock_fetch_and_cache, + patch("weko_group_cache_db.groups.time"), + patch("weko_group_cache_db.groups.traceback") as mock_traceback, + progress_signal.connected_to(mock_receiver), + ): + mock_func = MagicMock(side_effect=[2, 3]) + mock_fetch_and_cache.return_value = mock_func + + fetch_all(toml_path="institutions.toml") + + assert mock_func.call_count == num_institutions + assert mock_receiver.call_count == num_institutions + 1 + assert mock_traceback.print_exc.call_count == num_institutions + 1 + assert log_capture.records[0].getMessage() == f"Successfully cached 2 groups for {data[0]['fqdn']}." + assert log_capture.records[1].getMessage() == f"Successfully cached 3 groups for {data[1]['fqdn']}." + + +def test_fetch_and_cache_executed_signal_receiver_error(institutions_data, set_test_config): + data = institutions_data(1) + institution = Institution(**data[0]) + set_test_config() + groups = ["jc_group1", "jc_group2"] + + mock_store = MagicMock() + mock_receiver = MagicMock(side_effect=Exception("Receiver error")) + with ( + patch("weko_group_cache_db.groups.fetch_map_groups") as mock_fetch, + patch("weko_group_cache_db.groups.set_groups_to_redis") as mock_set, + patch("weko_group_cache_db.groups.traceback") as mock_traceback, + executed_signal.connected_to(mock_receiver), + ): + mock_fetch.return_value = groups + mock_set.return_value = datetime(2025, 1, 1, tzinfo=UTC) + + func = fetch_and_cache() + count = func(institution, mock_store) + + assert count == len(groups) + mock_fetch.assert_called_once_with(institution) + mock_set.assert_called_once_with(institution.fqdn, groups, store=mock_store) + mock_receiver.assert_called_once() + mock_traceback.print_exc.assert_called_once() diff --git a/tests/test_redis.py b/tests/test_redis.py index 55299a8..8b3c03b 100644 --- a/tests/test_redis.py +++ b/tests/test_redis.py @@ -16,7 +16,7 @@ # def connection() -> Redis: def test_connection_redis(set_test_config, log_capture: pytest.LogCaptureFixture): - set_test_config(REDIS_TYPE="redis") + set_test_config(REDIS_TYPE="RedisCache") with patch("weko_group_cache_db.redis._redis_connection") as mock_redis_connection: mock_redis_connection.return_value = MagicMock(spec=redis.Redis) store = connection() @@ -28,7 +28,7 @@ def test_connection_redis(set_test_config, log_capture: pytest.LogCaptureFixture def test_connection_sentinel(set_test_config, log_capture: pytest.LogCaptureFixture): - set_test_config(REDIS_TYPE="sentinel") + set_test_config(REDIS_TYPE="RedisSentinelCache") with patch("weko_group_cache_db.redis._sentinel_connection") as mock_sentinel_connection: mock_sentinel_connection.return_value = MagicMock(spec=redis.Redis) store = connection() @@ -40,7 +40,7 @@ def test_connection_sentinel(set_test_config, log_capture: pytest.LogCaptureFixt def test_connection_catch_value_error(set_test_config, log_capture: pytest.LogCaptureFixture): - set_test_config(REDIS_TYPE="redis") + set_test_config(REDIS_TYPE="RedisCache") with patch("weko_group_cache_db.redis._redis_connection") as mock_redis_connection: mock_redis_connection.side_effect = ValueError("Test ValueError") with pytest.raises(ValueError, match="Test ValueError"): @@ -50,7 +50,7 @@ def test_connection_catch_value_error(set_test_config, log_capture: pytest.LogCa def test_connection_catch_connection_error(set_test_config, log_capture: pytest.LogCaptureFixture): - set_test_config(REDIS_TYPE="redis") + set_test_config(REDIS_TYPE="RedisCache") with patch("weko_group_cache_db.redis._redis_connection") as mock_redis_connection: mock_store = MagicMock(spec=redis.Redis) mock_store.ping.side_effect = RedisConnectionError("Test ConnectionError") diff --git a/uv.lock b/uv.lock index 4fd256c..cfb9afb 100644 --- a/uv.lock +++ b/uv.lock @@ -20,6 +20,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/df/73/b6e24bd22e6720ca8ee9a85a0c4a2971af8497d8f3193fa05390cbd46e09/backoff-2.2.1-py3-none-any.whl", hash = "sha256:63579f9a0628e06278f7e47b7d7d5b6ce20dc65c5e96a6f3ca99a6adca0396e8", size = 15148, upload-time = "2022-10-05T19:19:30.546Z" }, ] +[[package]] +name = "blinker" +version = "1.9.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/21/28/9b3f50ce0e048515135495f198351908d99540d69bfdc8c1d15b73dc55ce/blinker-1.9.0.tar.gz", hash = "sha256:b4ce2265a7abece45e7cc896e98dbebe6cead56bcf805a3d23136d145f5445bf", size = 22460, upload-time = "2024-11-08T17:25:47.436Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/10/cb/f2ad4230dc2eb1a74edf38f1a38b9b52277f75bef262d8908e60d957e13c/blinker-1.9.0-py3-none-any.whl", hash = "sha256:ba0efaa9080b619ff2f3459d1d500c57bddea4a6b424b60a91141db6fd2f08bc", size = 8458, upload-time = "2024-11-08T17:25:46.184Z" }, +] + [[package]] name = "certifi" version = "2025.10.5" @@ -516,10 +525,11 @@ wheels = [ [[package]] name = "weko-group-cache-db" -version = "1.0.0rc4" -source = { virtual = "." } +version = "1.0.0rc6" +source = { editable = "." } dependencies = [ { name = "backoff" }, + { name = "blinker" }, { name = "inflect" }, { name = "pydantic-settings" }, { name = "redis" }, @@ -540,9 +550,10 @@ dev = [ [package.metadata] requires-dist = [ { name = "backoff", specifier = ">=2.2.1" }, + { name = "blinker", specifier = ">=1.9.0" }, { name = "inflect", specifier = ">=7.5.0" }, { name = "pydantic-settings", specifier = ">=2.12.0" }, - { name = "redis", specifier = ">=7.0.1" }, + { name = "redis", specifier = ">=6.4.0" }, { name = "requests", specifier = ">=2.32.5" }, { name = "rich-click", specifier = ">=1.9.4" }, { name = "werkzeug", specifier = ">=3.1.3" }, diff --git a/weko_group_cache_db/__init__.py b/weko_group_cache_db/__init__.py index 70e895d..b537b8f 100644 --- a/weko_group_cache_db/__init__.py +++ b/weko_group_cache_db/__init__.py @@ -4,4 +4,4 @@ """weko-group-cache-db is a caching solution for group information in Redis.""" -__version__ = "1.0.0rc4" +__version__ = "1.0.0rc6" diff --git a/weko_group_cache_db/config.py b/weko_group_cache_db/config.py index af5c5a6..101b957 100644 --- a/weko_group_cache_db/config.py +++ b/weko_group_cache_db/config.py @@ -4,25 +4,21 @@ """Settings module for weko-group-cache-db.""" -import tomllib import typing as t from contextvars import ContextVar from pathlib import Path - -import rich_click as click +from typing import overload from pydantic import BaseModel, computed_field from pydantic_settings import ( BaseSettings, PydanticBaseSettingsSource, SettingsConfigDict, + TomlConfigSettingsSource, ) from werkzeug.local import LocalProxy -if t.TYPE_CHECKING: - from pydantic.fields import FieldInfo # pragma: no cover - class Settings(BaseSettings): """Settings for application with validation.""" @@ -48,7 +44,7 @@ class Settings(BaseSettings): If it specified less than 0, it will be considered as no expiration. """ - MAP_GROUPS_API_ENDPOINT: str + MAP_GROUPS_API_ENDPOINT: str = "https://sample.gakunin.jp/api/groups/" """Map groups API endpoint.""" REQUEST_TIMEOUT: t.Annotated[int, "seconds"] = 20 @@ -69,8 +65,8 @@ class Settings(BaseSettings): REQUEST_RETRY_MAX: t.Annotated[int | float, "seconds"] = 90 """Maximum time for exponential backoff during request retries.""" - REDIS_TYPE: t.Literal["redis", "sentinel"] = "redis" - """Redis type to use. `redis` or `sentinel` is allowed.""" + REDIS_TYPE: t.Literal["RedisCache", "RedisSentinelCache"] = "RedisCache" + """Redis type to use. `RedisCache` or `RedisSentinelCache` is allowed.""" REDIS_HOST: str = "localhost" """Redis service host name.""" @@ -83,7 +79,7 @@ class Settings(BaseSettings): @computed_field @property - def REDIS_URL(self) -> str: # noqa: N802 + def REDIS_URL(self) -> str: # ruff: ignore[invalid-function-name] """Redis URL for caching group information.""" return f"redis://{self.REDIS_HOST}:{self.REDIS_PORT}/{self.REDIS_DB_INDEX}" @@ -95,7 +91,7 @@ def REDIS_URL(self) -> str: # noqa: N802 @computed_field @property - def REDIS_SENTINELS(self) -> list[tuple[str, str]]: # noqa: N802 + def REDIS_SENTINELS(self) -> list[tuple[str, str]]: # ruff: ignore[invalid-function-name] """A list of Redis sentinel host names and their ports.""" return ( [(sentinel.host, str(sentinel.port)) for sentinel in self.SENTINELS] @@ -106,7 +102,10 @@ def REDIS_SENTINELS(self) -> list[tuple[str, str]]: # noqa: N802 model_config = SettingsConfigDict( extra="forbid", frozen=True, + alias_generator=lambda s: s.lower(), validate_default=True, + validate_by_name=True, + validate_by_alias=True, ) @classmethod @@ -158,44 +157,26 @@ class Sentinel(BaseModel): """Sentinel port number.""" -class TomlConfigSettingsSource(PydanticBaseSettingsSource): - """TOML configuration settings source.""" - - def __init__(self, settings_cls: type[BaseSettings], toml_path: Path) -> None: - """Initialize TOML config settings source.""" - super().__init__(settings_cls) - self.toml_path = toml_path - - def get_field_value( # noqa: D102 - self, field: FieldInfo, field_name: str - ) -> tuple[t.Any, str, bool]: ... - - def __call__(self) -> dict[str, t.Any]: - """Load settings from TOML file.""" # noqa: DOC201 - if not self.toml_path.exists(): - click.echo("[WARN] Settings file not found. Default values will be used.") - click.echo(f"[INFO] Looking for settings file at: {self.toml_path}") - return {} - - with self.toml_path.open("rb") as f: - data = tomllib.load(f) - - return {k.upper(): v for k, v in data.items()} - - def __repr__(self) -> str: - """Representation.""" # noqa: DOC201 - return ( - f"TomlConfigSettingsSource(toml_path={self.toml_path})" # pragma: no cover - ) - - _no_config_msg = "Config has not been initialized." _current_config: ContextVar[Settings] = ContextVar("current_config") -def setup_config(toml_path: str) -> None: +@overload +def setup_config(config: dict[str, t.Any]) -> None: ... +@overload +def setup_config(config: str) -> None: ... +@overload +def setup_config(config: Settings) -> None: ... + + +def setup_config(config: dict[str, t.Any] | str | Settings) -> None: """Initialize the global config instance.""" - _current_config.set(Settings(toml_path=toml_path)) # pyright: ignore[reportCallIssue] + if isinstance(config, dict): + _current_config.set(Settings(**config)) # pyright: ignore[reportCallIssue] + elif isinstance(config, str): + _current_config.set(Settings(toml_path=config)) # pyright: ignore[reportCallIssue] + else: + _current_config.set(config) config = t.cast(Settings, LocalProxy(_current_config, unbound_message=_no_config_msg)) diff --git a/weko_group_cache_db/exc.py b/weko_group_cache_db/exc.py index 96d097c..08583fa 100644 --- a/weko_group_cache_db/exc.py +++ b/weko_group_cache_db/exc.py @@ -24,7 +24,7 @@ def __init__(self, fqdn: str, origin: Exception | None = None) -> None: self.fqdn = fqdn self.origin = origin - def __str__(self) -> str: # noqa: D105 + def __str__(self) -> str: # ruff: ignore[undocumented-magic-method] return f"FQDN: {self.fqdn}, {self.origin}" diff --git a/weko_group_cache_db/groups.py b/weko_group_cache_db/groups.py index bfd9ec5..b2ad2c9 100644 --- a/weko_group_cache_db/groups.py +++ b/weko_group_cache_db/groups.py @@ -22,9 +22,11 @@ from .loader import Institution, InstitutionSource, load_institutions from .logger import console, logger from .redis import connection +from .signals import ExecutedData, ProgressData, executed_signal, progress_signal if t.TYPE_CHECKING: - from redis import Redis # pragma: no cover + from backoff.types import Details + from redis import Redis def fetch_all(**kwargs: t.Unpack[InstitutionSource]): @@ -61,6 +63,8 @@ def fetch_all(**kwargs: t.Unpack[InstitutionSource]): ) for index, institution in enumerate(institutions): + _send_progress_signal(institutions, index) + try: group_count = fetch_and_cache()(institution, store) logger.info( @@ -81,6 +85,8 @@ def fetch_all(**kwargs: t.Unpack[InstitutionSource]): if index != total - 1: time.sleep(config.REQUEST_INTERVAL) + _send_progress_signal(institutions, total) + if exceptions: error_message = "Failed to update information from %d institution(s)." logger.error(error_message, len(exceptions)) @@ -137,6 +143,16 @@ def fetch_and_cache(): """ + def _count(details: Details) -> None: + """Update the retry count based on backoff details. + + Arguments: + details (Details): Details from backoff. + + """ + nonlocal retries + retries = details.get("tries", 0) + @backoff.on_exception( lambda: backoff.expo( base=config.REQUEST_RETRY_BASE, @@ -146,25 +162,33 @@ def fetch_and_cache(): (requests.RequestException, redis.RedisError), max_tries=config.REQUEST_RETRIES + 1, jitter=backoff.full_jitter, + on_backoff=_count, ) def _retrieve_fetch_and_cache(institution: Institution, store: Redis) -> int: + nonlocal retries try: group_ids = fetch_map_groups(institution) - set_groups_to_redis(institution.fqdn, group_ids, store=store) + updated_at = set_groups_to_redis(institution.fqdn, group_ids, store=store) + _send_executed_signal( + institution, "success", retries=retries, updated_at=updated_at + ) return len(group_ids) - except requests.RequestException: + except requests.RequestException as ex: logger.warning( "Failed to fetch groups from mAP API for institution: %s", institution.fqdn, ) + _send_executed_signal(institution, "failed", retries=retries, error=ex) raise - except redis.RedisError: + except redis.RedisError as ex: logger.warning( "Failed to cache groups to Redis for institution: %s", institution.fqdn, ) + _send_executed_signal(institution, "failed", retries=retries, error=ex) raise + retries = 0 return _retrieve_fetch_and_cache @@ -195,7 +219,9 @@ def fetch_map_groups(institution: Institution) -> list[str]: return group_ids -def set_groups_to_redis(fqdn: str, group_ids: list[str], *, store: Redis | None = None): +def set_groups_to_redis( + fqdn: str, group_ids: list[str], *, store: Redis | None = None +) -> datetime: """Set groups to redis. Arguments: @@ -204,10 +230,13 @@ def set_groups_to_redis(fqdn: str, group_ids: list[str], *, store: Redis | None store(Redis | None): Redis store object. If None, a new connection will be established. + Returns: + datetime: The timestamp when the groups were cached. + """ - transformed_fqdn = fqdn.replace(".", "_").replace("-", "_") - redis_key = transformed_fqdn + config.CACHE_KEY_SUFFIX - updated_at = datetime.now(UTC).isoformat(timespec="seconds") + redis_key = cache_key(fqdn) + now = datetime.now(UTC) + updated_at = now.isoformat(timespec="seconds") if store is None: store = connection() @@ -218,3 +247,59 @@ def set_groups_to_redis(fqdn: str, group_ids: list[str], *, store: Redis | None store.persist(redis_key) if config.CACHE_TTL >= 0: store.expire(redis_key, config.CACHE_TTL) + + return now + + +def cache_key(fqdn: str) -> str: + """Generate the Redis key for the given FQDN. + + Arguments: + fqdn (str): FQDN of the institution. + + Returns: + str: Redis key for the institution's groups. + + """ + transformed_fqdn = fqdn.replace(".", "_").replace("-", "_") + return transformed_fqdn + config.CACHE_KEY_SUFFIX + + +def _send_progress_signal(institutions: list[Institution], index: int) -> None: + total = len(institutions) + if index >= total: + status = "completed" + elif index > 0: + status = "in_progress" + else: + status = "started" + current = institutions[index].fqdn if index < total else "N/A" + + data = ProgressData(status=status, total=total, done=index, current=current) + + try: + progress_signal.send(fetch_all, data=data) + except Exception: # ruff: ignore[blind-except] + traceback.print_exc() + + +def _send_executed_signal( + institution: Institution, + status: t.Literal["success", "failed"], + updated_at: datetime | None = None, + retries: int = 0, + error: Exception | None = None, +) -> None: + data = ExecutedData( + fqdn=institution.fqdn, + status=status, + retries=retries, + error_type=type(error).__name__ if error else None, + error_message=str(error) if error else None, + updated_at=updated_at or datetime.now(UTC), + ) + + try: + executed_signal.send(fetch_all, data=data) + except Exception: # ruff: ignore[blind-except] + traceback.print_exc() diff --git a/weko_group_cache_db/loader.py b/weko_group_cache_db/loader.py index d66220d..b342d54 100644 --- a/weko_group_cache_db/loader.py +++ b/weko_group_cache_db/loader.py @@ -34,6 +34,9 @@ class InstitutionSource(t.TypedDict): fqdn_list_file: t.NotRequired[str | Path] """Path to the file containing FQDN list.""" + fqdn_list: t.NotRequired[list[str]] + """List of FQDNs.""" + def load_institutions(**kwargs: t.Unpack[InstitutionSource]) -> list[Institution]: """Load institution information from the configured source. @@ -52,6 +55,10 @@ def load_institutions(**kwargs: t.Unpack[InstitutionSource]) -> list[Institution return load_institutions_from_directory( kwargs["directory_path"], kwargs["fqdn_list_file"] ) + if "directory_path" in kwargs and "fqdn_list" in kwargs: + return load_institutions_from_list( + kwargs["directory_path"], kwargs["fqdn_list"] + ) logger.error("Invalid institution source configuration.") return [] @@ -157,8 +164,6 @@ def load_institutions_from_directory( list[Institution]: List of Institution objects. """ - if isinstance(directory_path, str): - directory_path = Path(directory_path) if isinstance(fqdn_list_file, str): fqdn_list_file = Path(fqdn_list_file) @@ -168,6 +173,25 @@ def load_institutions_from_directory( for line in f if (L := line.strip()) and not L.startswith("#") ] + return load_institutions_from_list(directory_path, fqdn_list) + + +def load_institutions_from_list( + directory_path: str | Path, fqdn_list: list[str] +) -> list[Institution]: + """Load institution information from the provided FQDN list. + + Arguments: + directory_path (str | Path): Path to the directory containing TOML files. + fqdn_list (list[str]): List of FQDNs. + + Returns: + list[Institution]: List of Institution objects. + + """ + if isinstance(directory_path, str): + directory_path = Path(directory_path) + p = inflect.engine() institutions: list[Institution] = [] with Progress( @@ -240,7 +264,7 @@ def check_existence_file(institution: Institution) -> bool: class Institution(BaseModel): """Institution model for loading from TOML file.""" - name: str | None + name: str | None = None """Name of the institution. Optional.""" fqdn: str diff --git a/weko_group_cache_db/logger.py b/weko_group_cache_db/logger.py index 3501124..2eba9c8 100644 --- a/weko_group_cache_db/logger.py +++ b/weko_group_cache_db/logger.py @@ -17,7 +17,7 @@ from .config import config if t.TYPE_CHECKING: - from datetime import datetime # pragma: no cover + from datetime import datetime def _log_time_format(datetime: datetime) -> Text: diff --git a/weko_group_cache_db/redis.py b/weko_group_cache_db/redis.py index 3ad30b3..6e77173 100644 --- a/weko_group_cache_db/redis.py +++ b/weko_group_cache_db/redis.py @@ -22,8 +22,8 @@ def connection() -> Redis: ConnectionError: If failed to connect to Redis. """ - try: - if config.REDIS_TYPE == "redis": + try: # ruff: ignore[too-many-statements-in-try-clause] + if config.REDIS_TYPE == "RedisCache": store = _redis_connection() store.ping() logger.info("Successfully connected to Redis.") diff --git a/weko_group_cache_db/signals.py b/weko_group_cache_db/signals.py new file mode 100644 index 0000000..5797c40 --- /dev/null +++ b/weko_group_cache_db/signals.py @@ -0,0 +1,58 @@ +# +# Copyright (C) 2025 National Institute of Informatics. +# + +"""Signal definitions for weko-group-cache-db.""" + +import typing as t + +from datetime import datetime # ruff: ignore[typing-only-standard-library-import] + +from blinker import Namespace +from pydantic import BaseModel + +namespace = Namespace() + +# Define a custom signal for updating count +progress_signal = namespace.signal("progress") + +# Define a custom signal for updating result +executed_signal = namespace.signal("executed") + + +class ProgressData(BaseModel): + """Data model for progress signal.""" + + status: t.Literal["started", "in_progress", "completed"] + """Status of the fetch and cache operation.""" + + total: int + """Total number of institutions to process.""" + + done: int + """Number of institutions processed so far.""" + + current: str + """FQDN of the current institution being processed.""" + + +class ExecutedData(BaseModel): + """Data model for executed signal.""" + + fqdn: str + """FQDN of the institution.""" + + status: t.Literal["success", "failed"] + """Status of the fetch and cache operation.""" + + retries: int = 0 + """Number of retries attempted.""" + + error_type: str | None = None + """Type of error, if any.""" + + error_message: str | None = None + """Error message, if any.""" + + updated_at: datetime + """Timestamp of the last update."""