CLAUDE.md

This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.

What this is

Temporal.io worker for industrial time-series data ingestion (scouter). Pulls data from MongoDB / PI Web API, runs quality-gate filtering + aggregation, writes to PostgreSQL, caches state in Redis. Heavy dependency on the private sientia_do library (aka sientia-dataops-library) for base activity classes (Postgres/Redis/Mongo repos), logging, notifications, metrics, retry policy, and connector config builders — read that lib's source when a base method/behavior isn't in this repo.

Setup

python3.11 -m venv venv
source ./venv/bin/activate
pip install -r requirements-local.txt   # uses git+ssh to sientia-dataops-library (needs SSH access)
pip install -r requirements-dev.txt     # lint/test/e2e tooling
cp .env.example .env                    # fill in connection details

requirements.txt uses the plain package name sientia_do (resolved via index/private registry in CI); requirements-local.txt pins it via git+ssh://...sientia-dataops-library.git@<tag> for local dev — keep both in sync when bumping the library version (see git-requirements-mapping.txt).

Commands

./run_local.sh                          # activate venv, load .env, run worker
python -m scouter.worker.worker          # run worker manually

pytest                                   # unit tests (tests/), Docker-free
pytest tests/activities/test_redis.py    # single file
pytest tests/workflow/test_scouter.py::TestClassName::test_name  # single test
pytest --cov=scouter --cov-report=html   # coverage (or ./run_coverage.sh, opens browser)

# e2e tests are excluded from default `pytest` runs (testpaths=tests) — invoke explicitly:
pytest e2e/ --override-ini testpaths=e2e -m e2e -v
pytest e2e/ --override-ini testpaths=e2e -m e2e -x    # stop on first failure

ruff check .                             # lint
ruff format .                             # format
mypy scouter                             # type check
bandit -r scouter                        # security scan

CI quality gate (.github/workflows/quality-gate.yml) calls a shared reusable workflow (Aignosi/github_workflow_templates) with requirements_file: requirements-local.txt; releases (release.yml) fire on merged PRs to main via another shared workflow. Both are effectively opaque here — the actual lint/test/coverage steps live in the github_workflow_templates repo.

Architecture

Parent/child Temporal workflow pattern, two independent worker task queues sharing the same CoreScouter sub-workflow:

Scouter (scouter/workflow/scouter.py)              PIWebAPIScouter (scouter/workflow/pi_web_api_scouter.py)
  reads raw_{schedule_name} from MongoDB              queries PI Web API directly (get_tag_values)
  tracks last-processed timestamp in Redis             normalizes all timestamps to the batch max
        │                                                     │
        └───────────────► CoreScouter (scouter/workflow/sub_workflows/core_scouter.py) ◄───────┘
                              1. data_quality_gate   (NULL_VALUES_FILTER / OUT_OF_BOUNDS_FILTER, DISCARD policy)
                              2. aggregate_data      (per-tag: lts/avg/mdn/max/min)
                              3. group_and_hold_data (Redis, TTL = retention_time)
                              4. export_data_to_postgres
                              5. write_metrics       (Prometheus)
                              6. store_data_package  (optional, MongoDB, if debug_data_package=true)

Activities (scouter/activities/activities.py) is a multiple-inheritance god-class combining Postgres (from sientia_do), Redis, Gates, MongoDB, API (this repo's scouter/activities/*.py). Each mixin owns its own __init__/close — the worker calls all of them explicitly. When adding a new activity, decide which mixin it belongs to; don't bypass the split.
Temporal sandboxing: every module doing non-workflow-safe imports (pandas, sientia_do, os, etc.) wraps them in with workflow.unsafe.imports_passed_through():. Follow this pattern in any new workflow/activity file — Temporal's sandbox will fail otherwise.
Two task queues, one worker process: worker.py builds two prepare_worker(...) instances (main workflow Scouter vs PIWebAPIScouter, both with CoreScouter as a registered child) and runs them concurrently via asyncio.gather. If either crashes, the whole process exits non-zero (see the finally block) — that's deliberate, for k8s restart semantics.
Aggregation functions: lts (latest), avg, mdn (median), max, min — configured per-tag in model_tags[tag].aggr_function.
Quality filters: NULL_VALUES_FILTER, OUT_OF_BOUNDS_FILTER (scouter/utils/quality/filters.py), each with a configurable policy (currently DISCARD). Register new filters in the quality_gate_filters dict in scouter/activities/gates.py.
Config builders (build_postgres_config, build_redis_config, build_mongodb_config, build_api_config) come from sientia_do.utils.connectors_config, not this repo — scouter/utils/connectors_config.py only has build_kafka_config locally.
Metrics (scouter/metrics.py): APP_UP gauge, LABORIOUS_DATA_WRITTEN_COUNT counter, TAG_CHANGES_MONITOR gauge, all labeled by pod_id/model_name/workflow_name (+tag_name for the last one). sonar-project.properties excludes scouter/worker/worker.py from coverage — it's the process entrypoint, exercised by e2e, not unit tests.

Testing structure

tests/ — Docker-free unit tests, mirrors scouter/ package layout (tests/activities/, tests/workflow/, tests/utils/). Default pytest invocation only runs this dir (testpaths=tests in pyproject.toml).
e2e/ — production-faithful tests against real testcontainers (Postgres/Mongo/Redis) + WorkflowEnvironment.start_local() + an in-process PI Web API test server (e2e/pi_web_api_test_server.py). Only Logger and an optional notification-insert spy are mocked; everything else is real Activities wiring. Must be invoked explicitly (see Commands) — not picked up by default pytest.
- Scenario catalog lives in e2e/scenarios.md, numbered and mapped to test_scenario_* functions. Section ## 0 = harness smoke tests (e2e/test_harness_smoke.py) that check Docker/container/worker wiring liveness before the real scenarios run.
- A scenario that fails due to a genuine production defect gets marked xfail(strict=True) rather than deleted/skipped silently.
Markers: asyncio, e2e, integration, unit (declared in pyproject.toml).

Root-level scripts

get_data_pims.py, pi_web_api_fetch_data.py, init_port_forward.sh are standalone helper/debug scripts (PI/PIMS data fetch, k8s port-forwarding for local dev against a real cluster) — not part of the worker's runtime path.

6.9 KiB Raw Blame History