# CLAUDE.md

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

## What This Is

Orchard Simple Workflow Feed Ingestion — a Python monorepo containing 80+ ETL workflows that ingest data from music industry partners (Spotify, Apple Music, Deezer, YouTube, Pandora, TikTok, etc.) into Snowflake. Built on Amazon SWF orchestrated via the Garcon workflow framework. Owned by `@theorchard/data-platform`.

After fact table loading, many flows trigger downstream dbt builds via Jenkins (e.g., `dbt-scheduler-analytics-pipeline`, `dbt-scheduler-analytics-parametrized-pipeline`, `dbt-scheduler-analytics-build-views`) in the separate `dbt-analytics` repo.

## Common Commands

```sh
# Dev setup
cp .env.shadow .env                                # fill in credentials (AWS, Snowflake, etc.)
make install_from_lock_include_dev                  # uv sync --locked --dev + uv pip install -e .

# Unit tests
make test_unit                                     # run all
make test_unit_path TEST_PATH=tests/flows/deezer   # single flow (-x -vv)
uv run py.test tests/flows/deezer/test_tasks.py::TestDownload::test_download -x -vv  # single test

# Lint (uses --no-env-file to prevent .env from polluting lint runs)
make lint         # flake8 on feed_ingestion/ and tests/
make check_deps   # deptry (unused/missing dependency check)
make check_lock   # verify uv.lock matches pyproject.toml

# Docker
make docker_login       # ECR login (prod account 086679231553)
make docker_login_dev   # ECR login (dev account 103233932089)
make docker_build       # docker compose build
make docker_unit_lint   # full test + lint in Docker (used by CI)
make docker_scan        # local CVE vulnerability scan (run docker_build first)

# Run a flow locally (via docker compose)
docker compose up                       # starts decider + worker
docker compose exec worker garcon exec <flow_name> -c '{"context_date": "2024-06-16"}'

# Run a flow locally (via uv)
uv run garcon decider <flow_name>
uv run garcon worker <flow_name>
uv run garcon exec <flow_name> -c '{"context_date": "2024-06-16"}'

# Local activity execution without SWF (via garcon-contrib garcon-activity-local CLI)
# See: https://github.com/theorchard/garcon-contrib/tree/master/contrib-cli
feed_ingestion.flows.<flow_name>.flow -c Flow list
feed_ingestion.flows.<flow_name>.flow -c Flow run <activity_name> -cf ./context.json
```

## Architecture

### Workflow Engine

Each ETL flow runs as three Amazon SWF processes: **decider** (polls for decision tasks, schedules activities per the DAG), **worker** (polls for activity tasks, executes them), and **exec** (triggers a new workflow execution with initial context JSON). In production, Supervisord manages 1 decider + 5 workers by default (`conf/supervisor.conf`), though some flows have custom supervisor configs with different counts (e.g., `conf/qq-supervisor.conf`).

### Three-Stage ETL Pattern

1. **Download** — fetch raw files from partner (SFTP/HTTP/S3 drop bucket/Apple Reporter JAR/Google API) → archive to `s3://[dev-]cucumbers/<FeedName>/archives/<date>/`
2. **Load Staging Raw** — denormalize files, COPY into Snowflake `staging_raw_<flow_name>` table
3. **Load Fact Tables** — JOIN staging raw against dimension tables → `fact_analytics` + `fact_analytics_error`

### Flow Package Structure

Each flow in `feed_ingestion/flows/<flow_name>/` typically contains:
- `flow.py` — Flow class composed from base classes/mixins, defines activity DAG via `schedule()`
- `exec.py` — defines how `garcon exec` launches this workflow
- `config.py` — flow-specific constants (S3 paths, table names, version rules)
- `tasks.py` — Garcon tasks decorated with `@task.decorate()`
- `snowflake_executor.py` — Snowflake SQL executor class (not all flows have this)
- `queries/` — SQL files for Snowflake COPY, MERGE operations (not all flows have this)

### Base Class Hierarchy (`feed_ingestion/flows/base.py`)

- `FlowBase` — SWF client setup, domain resolution, exception handler (Sentry)
- `FlowLicensor(FlowBase)` — extends FlowBase with multi-licensor support (theorchard, sme, altafonte, etc.)
- `FlowConfigMixin` — config loading mixin
- `FlowLoadRawMixinSF` — standardized staging_raw table loading
- `FlowLoadFactMixinSF` — standardized fact_analytics loading
- `FlowLoadMarketshareMixinSF` — market share table loading (used by deezer_marketshare, etc.)
- `FlowYouTubeMixin` — YouTube-specific loading logic

### Context Passing

Garcon passes a mutable `context` dict through all activities. Tasks use namespace prefixes (`bootstrap.date`, `bootstrap.feed_name`) and declare inputs via `.fill()` + namespace mapping. Keep contexts small — load config from `config.py` within tasks instead.

### Feed Status Tracking

DynamoDB table `{env}_feed_ingestion_status` (via `garcon_contrib.dynamo_feed_status`) tracks workflow state: `SOURCE_NOT_AVAIL` → `SOURCE_AVAILABLE` → `SOURCE_DOWNLOADED` → `POPULATED_RAW_TABLE` → `INGESTED`. Checked at bootstrap for idempotency — already-completed workflows skip re-processing.

### Secrets

AWS Secrets Manager via `secrets_manager.swf_ext.SWFSecretsManager`. Each flow has a `secrets_path` (e.g., `'deezer_daily'` in prod, `'swf-deezer'` in dev).

### S3 Layout

```
s3://[dev-]cucumbers/<FeedName>/
  archives/<YYYY-MM-DD>/                    # raw files from partner
  temp_staging_raw_<flow_name>/<YYYY-MM-DD>/  # intermediate staging data
  staging_raw_<flow_name>/<YYYY-MM-DD>/       # data loaded into staging_raw table
  fact_analytics/<YYYY-MM-DD>/              # data loaded into fact_analytics
  fact_analytics_error/<YYYY-MM-DD>/
```

Drop bucket: `s3://prod-orcdbucket/feed-drop/<FeedName>/` (prod), `s3://dev-feed-drop/<FeedName>/` (dev).

## Shared Code Layout

- `feed_ingestion/common/` — shared modules: `base_executor`, `staging_raw_sf`, `marketshare_sf`, `status_runner`
- `feed_ingestion/tasks/` — generic tasks used across flows (date, notification, load fact/raw, overall status)
- `feed_ingestion/util/` — utility modules: `aws/`, `jenkins/`, `kafka/`, `neo4j/`, `snowflake/`, `sentry_util`
- `feed_ingestion/conf/config.py` — runtime global config (Snowflake params, AWS region, Boto3 config)

## Testing

### Test Structure

Tests mirror the source tree: `tests/flows/<flow_name>/`, `tests/tasks/`, `tests/util/`. Deprecated flow tests are excluded via `pytest.ini` (`norecursedirs = tests/flows/_deprecated`).

### Test Environment Setup (`tests/conftest.py`)

`pytest_sessionstart` strips all `.env.shadow` variables from the environment and sets `Environment=dev` before any tests run. `pytest_configure` sets `sys._called_from_test = True` and `SWITCHBOARD_CONSUMER_KAFKA_TOPIC=test_kafka_topic`.

### Key Fixtures (`tests/conftest.py`)

- `clear_os_environ` — clears environment, sets `FEED_INGESTION_TABLE='dev_feed_ingestion_status'`
- `frozen_time` — freezes to `2024-11-11` via freezegun, also mocks `time.sleep`
- `mock_aws` — mocks all AWS services via `moto.mock_aws()`
- `sf_config_mock` — mock Snowflake connection params dict
- `aws_config_mock` — mock AWS credentials dict
- `SubstringMatcher` — helper for asserting SQL calls by substring list

### Integration Tests

Ruby/RSpec (`integration_tests/spec/`) and Python (`integration_tests/python_integration_tests/`) test end-to-end against Snowflake via ODBC. `get_flows_to_test.py` selects tests based on `git diff`.

## Code Style

- **Flake8**: Google-style imports (`import-order-style=google`), single quotes enforced, Google-style docstrings required. Docstring/import-order rules relaxed in `tests/`.
- **Naming**: all file/folder names lowercase with `_` separator. Workflow name = folder name under `flows/`.
- **Dependencies**: use `~=` (compatible release) version specifiers. Pin to major by default (`~=1.2`). Exact pins (`==`) only temporarily with a comment. See `DEPENDENCIES_MANAGEMENT.md`.
- **Package manager**: `uv` with `pyproject.toml`. Private PyPI index at `https://pypi.theorchard.io/pypi` for internal packages (`owsrequest`, `garcon_contrib`, `secrets_manager`, `snowflake_connector_etl`).
- **Python configs only** — YAML configs are deprecated (since 2015). Flow constants go in `config.py`.

## Guidelines

* when adding a new file add it git with `git add -f`
* when creating a test, add new line separators between initialization, actual tested call, checks and validations
* it is recommended to use parametrized tests

## Workflow Conventions

- Workflows must be **idempotent by date** — use feed status to skip already-completed work, never back out and reload data unnecessarily.
- Partially run workflows should **resume from where they left off** using feed status.
- Always check [garcon-contrib](https://github.com/theorchard/garcon-contrib) for existing reusable tasks before writing new ones.
- SNS notifications at workflow end: `prod-swf-feed-ingestion` (prod), `dev-feed-ingestion` (dev).

## CI/CD

Jenkins pipeline (`Jenkinsfile`). Stages: unit tests + lint + SAST + SonarQube + compliance (parallel) → Docker ECR push → optional deploy to QA/Prod ECS Fargate. Slack to `#data-alerts` on regression. PR retrigger via comment: `retest this please`. 80+ selectable SWF services for deployment.

## PR Checklist

* PRs should follow `pull_request_template.md`: title as `ISSUE-CODE meaningful title`, deployment instructions, monitoring resources (Monty, Looker), and reviewer checklist covering code clarity, naming, test coverage, resource cleanup, and failure messaging.