# Architecture Bulk contract creation tool for the Abacus GraphQL API. Reads input files (CSV, JSON, XLSX), validates and transforms rows, creates contracts via GraphQL, and writes results. ## Layering ``` cli.py ──→ app.py ──→ ContractProcessor (domain/) │ │ │ ├──→ ContractInputBuilder (domain/) │ │ └──→ entity lookup maps │ │ │ ├──→ RowReader (infra/) │ │ │ │ │ └──→ FileReader (protocol) │ │ ├── CsvFileReader │ │ ├── JsonFileReader │ │ └── XlsxFileReader │ │ │ ├──→ AbacusClient (connectors/) │ │ └──→ GraphQLClient │ │ └──→ Throttler → TokenBucketStrategy │ │ │ └──→ OutputSink (protocol) │ ├── CsvOutputSink │ └── NullOutputSink │ ├──→ config.py ──→ Config + RunOptions │ └──→ load_processed_identifiers (resume) ``` **Rules:** - `cli.py` is the only module that calls `sys.exit` or configures logging - `app.py` wires dependencies and calls the processor — can be used from CLI, Lambda, or tests - `domain/processor.py` orchestrates but doesn't do I/O directly (delegates to injected services) - `connectors/` handles GraphQL access — nothing else imports `requests` - `infra/` handles file I/O, reading, rate limiting, and output sinks - `schemas/` is purely declarative (models, enums, columns) ## Data Flow ``` Input file (.csv, .json, .xlsx) │ ▼ RowReader._reader_for(path) → selects FileReader by extension │ ▼ FileReader.read() → yields list[str] rows (streaming) │ ▼ DictRowIterator → first row = headers, yields (idx, dict) │ ▼ ContractRowIterator → validates + coerces → (idx, ContractRow) │ ▼ ContractProcessor.process() │ ├── Check: duplicate in input file? → skip ├── Check: already processed (resume)? → skip │ ├── ContractInputBuilder.build(row, policy) │ ├── Validate required fields → BuildSkip on failure │ ├── Look up signing entity / run controller (dict maps) │ ├── Apply business defaults (tracked in defaults_applied) │ └── Return BuildSuccess(input, defaults_applied) │ ├── [dry_run?] → log and return GraphQLResult(status=DRY_RUN) │ └── [execute] → AbacusClient.create_contract_with_lifecycles() │ ├── GraphQLClient.query() (uses requests.Session) │ └── Throttler.acquire() (token bucket) │ └── Return GraphQLResult[AbacusContract] │ ▼ OutputSink.add_success(row) → buffered, auto-flushed OutputSink.write_failures() → at end if errors/skipped OutputSink.close() → flush remaining ``` ## Key Design Decisions ### Dry-run by default The CLI defaults to dry-run mode. `--execute` is required to make real API calls. Dry-run is a constructor parameter on `ContractProcessor` — the processor decides whether to call the API client or log and return a stub result. ### BuildResult discriminated union `ContractInputBuilder.build()` returns `BuildSuccess | BuildSkip`. `BuildSuccess` carries the GraphQL input and a tuple of `defaults_applied` strings, making it explicit when business defaults (today's date, 30-day termination, etc.) were substituted for missing data. `MissingFieldPolicy.SKIP` vs `.DEFAULT` controls whether missing optional fields cause the row to be skipped or defaulted. ### OutputSink protocol The processor delegates all file output to an injected `OutputSink`. `CsvOutputSink` buffers success rows and auto-flushes every N rows, determines append mode from file existence, and derives the failures CSV path from the success CSV path. `NullOutputSink` is used when no output is configured. The processor never opens files directly. ### File format detection `RowReader` accepts a dict of `{extension: FileReader}` at construction. It detects the file extension and selects the matching reader. The pipeline is: `FileReader` → `DictRowIterator` → `ContractRowIterator`. Adding a new format means implementing `FileReader.read()` and registering the extension. ### Constructor DI everywhere Every class receives its collaborators via constructor. Protocols (`FileReader`, `OutputSink`, `RowReaderProtocol`, `ThrottlerStrategy`) enable test doubles without monkeypatching. ### ContractRow as the parsing boundary Raw dicts are converted to typed `ContractRow` instances at the `ContractRowIterator` boundary. Pydantic `mode='before'` validators handle all string-to-type coercion (dates, booleans, enums, integers). Everything downstream works with typed fields. ### RunOptions resolves three-way priority `RunOptions.resolve(config, **cli_overrides)` encapsulates the CLI arg > config.json > built-in default resolution. `app.run()` accepts a `RunOptions` instance — no parameter explosion. ### Config resolution priority `CLI arg > config.json > built-in default`. All resolution uses `is not None` checks (not truthiness) to correctly handle zero values. ### Entity maps from GraphQL Signing entity and run controller maps are loaded from the Abacus GraphQL gateway at startup. `AbacusClient` paginates run controllers automatically. ### Error handling: raise vs. return `AbacusClient` uses two error strategies depending on whether a failure is fatal or per-item: - **Fatal methods propagate exceptions** — `get_reference_signing_entities()`, `get_run_controllers()`, and `attach_run_controller()` let `ApiError` / `GraphQLError` bubble up. These are startup or single-call operations where failure means the batch cannot proceed. - **Per-item methods return error results** — `create_contract_with_lifecycles()` catches `ApiError` / `GraphQLError` and returns `GraphQLResult(status=ERROR, error=...)`. Individual contract creation failures should be collected, not abort the batch. The rule: if the caller processes many items and a single failure should not stop the others, return an error result. If failure means the run is invalid, raise. ### Resume capability `load_processed_identifiers()` reads (account_id, contract_name) pairs from an existing output CSV. The processor skips rows whose identifiers are already in that set. The `CsvOutputSink` determines append mode from file existence — no separate flag needed. ### Throttler in GraphQLClient Rate limiting is the transport client's concern. `GraphQLClient` uses a `requests.Session` for connection reuse and calls `Throttler.acquire()` before every request. ## Package Structure ``` src/ ├── cli.py CLI entry point ├── app.py Composition root ├── config.py Config model + RunOptions ├── domain/ │ ├── processor.py Batch orchestrator │ └── contract_input_builder.py Row → GraphQL input ├── connectors/ │ ├── abacus.py GraphQL domain client │ ├── graphql.py Generic GraphQL HTTP client │ └── errors.py ApiError ├── infra/ │ ├── readers.py FileReader protocol + CSV/JSON/XLSX │ ├── row_reader.py Row pipeline + RowReaderProtocol │ ├── output.py OutputSink protocol + CSV sink │ ├── resume.py Resume state loading │ └── throttler.py Rate limiter + TokenBucketStrategy └── schemas/ ├── columns.py CSV columns, MissingFieldPolicy, enums ├── enums.py Alias maps, BooleanValue, normalize ├── graphql.py GraphQL input/output Pydantic models ├── rows.py ContractRow (parsed CSV row) └── results.py BuildResult, GraphQLResult, ProcessingResults ``` ## Testing - **220 tests** total (215 unit + 5 integration) - **Unit tests** (`tests/unit/`) mirror the `src/` directory structure - **Integration tests** (`tests/integration/`) exercise the full pipeline through `app.run()` with `pytest-httpserver` providing a mock GraphQL server - **FakeAbacusClient** (`tests/fakes.py`) satisfies the `AbacusClient` interface with per-method error injection - **`make test`** runs unit tests; **`make test_integration`** runs integration tests ## Docker Unit + lint + integration tests in Docker: ```bash make docker_test # unit + lint make docker_test_integration # integration tests make docker_down # tear down ``` ## Dependencies | Package | Purpose | |---------|---------| | pydantic | Config validation, payload models, CSV row parsing | | requests | HTTP client for GraphQL API | | openpyxl | XLSX file reading | | ijson | Streaming JSON parsing (optional, falls back to json.load) | | pytest | Unit + integration tests (dev) | | ruff | Linting + formatting (dev) | | pytest-httpserver | Mock HTTP server for tests (dev) |